<s> import os <EOL> import shutil <EOL> import sys <EOL> import tempfile <EOL> from unittest . mock import patch <EOL> from . . hashindex import NSIndex <EOL> from . . helpers import Location , IntegrityError <EOL> from . . locking import UpgradableLock , LockFailed <EOL> from . . remote import RemoteRepository , InvalidRPCMethod <EOL> from . . repository import Repository <EOL> from . import BaseTestCase <EOL> class RepositoryTestCaseBase ( BaseTestCase ) : <EOL> key_size = <NUM_LIT:32> <EOL> def open ( self , create = False ) : <EOL> return Repository ( os . path . join ( self . tmppath , '<STR_LIT>' ) , create = create ) <EOL> def setUp ( self ) : <EOL> self . tmppath = tempfile . mkdtemp ( ) <EOL> self . repository = self . open ( create = True ) <EOL> self . repository . __enter__ ( ) <EOL> def tearDown ( self ) : <EOL> self . repository . close ( ) <EOL> shutil . rmtree ( self . tmppath ) <EOL> def reopen ( self ) : <EOL> if self . repository : <EOL> self . repository . close ( ) <EOL> self . repository = self . open ( ) <EOL> class RepositoryTestCase ( RepositoryTestCaseBase ) : <EOL> def test1 ( self ) : <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> self . repository . put ( ( '<STR_LIT>' % x ) . encode ( '<STR_LIT:ascii>' ) , b'<STR_LIT>' ) <EOL> key50 = ( '<STR_LIT>' % <NUM_LIT:50> ) . encode ( '<STR_LIT:ascii>' ) <EOL> self . assert_equal ( self . repository . get ( key50 ) , b'<STR_LIT>' ) <EOL> self . repository . delete ( key50 ) <EOL> self . assert_raises ( Repository . ObjectNotFound , lambda : self . repository . get ( key50 ) ) <EOL> self . repository . commit ( ) <EOL> self . repository . close ( ) <EOL> with self . open ( ) as repository2 : <EOL> self . assert_raises ( Repository . ObjectNotFound , lambda : repository2 . get ( key50 ) ) <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> if x == <NUM_LIT:50> : <EOL> continue <EOL> self . assert_equal ( repository2 . get ( ( '<STR_LIT>' % x ) . encode ( '<STR_LIT:ascii>' ) ) , b'<STR_LIT>' ) <EOL> def test2 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . commit ( ) <EOL> self . repository . delete ( b'<STR_LIT>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:bar>' ) <EOL> self . repository . commit ( ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT:bar>' ) <EOL> def test_consistency ( self ) : <EOL> """<STR_LIT>""" <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT:foo>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:bar>' ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT:bar>' ) <EOL> self . repository . delete ( b'<STR_LIT>' ) <EOL> self . assert_raises ( Repository . ObjectNotFound , lambda : self . repository . get ( b'<STR_LIT>' ) ) <EOL> def test_consistency2 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT:foo>' ) <EOL> self . repository . commit ( ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT>' ) <EOL> self . repository . rollback ( ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT:foo>' ) <EOL> def test_overwrite_in_same_transaction ( self ) : <EOL> """<STR_LIT>""" <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . repository . commit ( ) <EOL> self . assert_equal ( self . repository . get ( b'<STR_LIT>' ) , b'<STR_LIT>' ) <EOL> def test_single_kind_transactions ( self ) : <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . commit ( ) <EOL> self . repository . close ( ) <EOL> self . repository = self . open ( ) <EOL> with self . repository : <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:bar>' ) <EOL> self . repository . commit ( ) <EOL> self . repository = self . open ( ) <EOL> with self . repository : <EOL> self . repository . delete ( b'<STR_LIT>' ) <EOL> self . repository . commit ( ) <EOL> def test_list ( self ) : <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> self . repository . put ( ( '<STR_LIT>' % x ) . encode ( '<STR_LIT:ascii>' ) , b'<STR_LIT>' ) <EOL> all = self . repository . list ( ) <EOL> self . assert_equal ( len ( all ) , <NUM_LIT:100> ) <EOL> first_half = self . repository . list ( limit = <NUM_LIT:50> ) <EOL> self . assert_equal ( len ( first_half ) , <NUM_LIT:50> ) <EOL> self . assert_equal ( first_half , all [ : <NUM_LIT:50> ] ) <EOL> second_half = self . repository . list ( marker = first_half [ - <NUM_LIT:1> ] ) <EOL> self . assert_equal ( len ( second_half ) , <NUM_LIT:50> ) <EOL> self . assert_equal ( second_half , all [ <NUM_LIT:50> : ] ) <EOL> self . assert_equal ( len ( self . repository . list ( limit = <NUM_LIT:50> ) ) , <NUM_LIT:50> ) <EOL> class RepositoryCommitTestCase ( RepositoryTestCaseBase ) : <EOL> def add_keys ( self ) : <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:bar>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:bar>' ) <EOL> self . repository . commit ( ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> self . repository . delete ( b'<STR_LIT>' ) <EOL> def test_replay_of_missing_index ( self ) : <EOL> self . add_keys ( ) <EOL> for name in os . listdir ( self . repository . path ) : <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> os . unlink ( os . path . join ( self . repository . path , name ) ) <EOL> self . reopen ( ) <EOL> with self . repository : <EOL> self . assert_equal ( len ( self . repository ) , <NUM_LIT:3> ) <EOL> self . assert_equal ( self . repository . check ( ) , True ) <EOL> def test_crash_before_compact_segments ( self ) : <EOL> self . add_keys ( ) <EOL> self . repository . compact_segments = None <EOL> try : <EOL> self . repository . commit ( ) <EOL> except TypeError : <EOL> pass <EOL> self . reopen ( ) <EOL> with self . repository : <EOL> self . assert_equal ( len ( self . repository ) , <NUM_LIT:3> ) <EOL> self . assert_equal ( self . repository . check ( ) , True ) <EOL> def test_replay_of_readonly_repository ( self ) : <EOL> self . add_keys ( ) <EOL> for name in os . listdir ( self . repository . path ) : <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> os . unlink ( os . path . join ( self . repository . path , name ) ) <EOL> with patch . object ( UpgradableLock , '<STR_LIT>' , side_effect = LockFailed ) as upgrade : <EOL> self . reopen ( ) <EOL> with self . repository : <EOL> self . assert_raises ( LockFailed , lambda : len ( self . repository ) ) <EOL> upgrade . assert_called_once_with ( ) <EOL> def test_crash_before_write_index ( self ) : <EOL> self . add_keys ( ) <EOL> self . repository . write_index = None <EOL> try : <EOL> self . repository . commit ( ) <EOL> except TypeError : <EOL> pass <EOL> self . reopen ( ) <EOL> with self . repository : <EOL> self . assert_equal ( len ( self . repository ) , <NUM_LIT:3> ) <EOL> self . assert_equal ( self . repository . check ( ) , True ) <EOL> def test_crash_before_deleting_compacted_segments ( self ) : <EOL> self . add_keys ( ) <EOL> self . repository . io . delete_segment = None <EOL> try : <EOL> self . repository . commit ( ) <EOL> except TypeError : <EOL> pass <EOL> self . reopen ( ) <EOL> with self . repository : <EOL> self . assert_equal ( len ( self . repository ) , <NUM_LIT:3> ) <EOL> self . assert_equal ( self . repository . check ( ) , True ) <EOL> self . assert_equal ( len ( self . repository ) , <NUM_LIT:3> ) <EOL> class RepositoryAppendOnlyTestCase ( RepositoryTestCaseBase ) : <EOL> def test_destroy_append_only ( self ) : <EOL> self . repository . append_only = True <EOL> with self . assert_raises ( ValueError ) : <EOL> self . repository . destroy ( ) <EOL> def test_append_only ( self ) : <EOL> def segments_in_repository ( ) : <EOL> return len ( list ( self . repository . io . segment_iterator ( ) ) ) <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . commit ( ) <EOL> self . repository . append_only = False <EOL> assert segments_in_repository ( ) == <NUM_LIT:1> <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . commit ( ) <EOL> assert segments_in_repository ( ) == <NUM_LIT:1> <EOL> self . repository . append_only = True <EOL> assert segments_in_repository ( ) == <NUM_LIT:1> <EOL> self . repository . put ( b'<STR_LIT>' , b'<STR_LIT:foo>' ) <EOL> self . repository . commit ( ) <EOL> assert segments_in_repository ( ) == <NUM_LIT:2> <EOL> class RepositoryCheckTestCase ( RepositoryTestCaseBase ) : <EOL> def list_indices ( self ) : <EOL> return [ name for name in os . listdir ( os . path . join ( self . tmppath , '<STR_LIT>' ) ) if name . startswith ( '<STR_LIT>' ) ] <EOL> def check ( self , repair = False , status = True ) : <EOL> self . assert_equal ( self . repository . check ( repair = repair ) , status ) <EOL> self . assert_equal ( [ name for name in os . listdir ( os . path . join ( self . tmppath , '<STR_LIT>' ) ) if '<STR_LIT>' in name ] , [ ] , '<STR_LIT>' ) <EOL> def get_objects ( self , * ids ) : <EOL> for id_ in ids : <EOL> self . repository . get ( ( '<STR_LIT>' % id_ ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> def add_objects ( self , segments ) : <EOL> for ids in segments : <EOL> for id_ in ids : <EOL> self . repository . put ( ( '<STR_LIT>' % id_ ) . encode ( '<STR_LIT:ascii>' ) , b'<STR_LIT:data>' ) <EOL> self . repository . commit ( ) <EOL> def get_head ( self ) : <EOL> return sorted ( int ( n ) for n in os . listdir ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT:data>' , '<STR_LIT:0>' ) ) if n . isdigit ( ) ) [ - <NUM_LIT:1> ] <EOL> def open_index ( self ) : <EOL> return NSIndex . read ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT>' . format ( self . get_head ( ) ) ) ) <EOL> def corrupt_object ( self , id_ ) : <EOL> idx = self . open_index ( ) <EOL> segment , offset = idx [ ( '<STR_LIT>' % id_ ) . encode ( '<STR_LIT:ascii>' ) ] <EOL> with open ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT:data>' , '<STR_LIT:0>' , str ( segment ) ) , '<STR_LIT>' ) as fd : <EOL> fd . seek ( offset ) <EOL> fd . write ( b'<STR_LIT>' ) <EOL> def delete_segment ( self , segment ) : <EOL> os . unlink ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT:data>' , '<STR_LIT:0>' , str ( segment ) ) ) <EOL> def delete_index ( self ) : <EOL> os . unlink ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT>' . format ( self . get_head ( ) ) ) ) <EOL> def rename_index ( self , new_name ) : <EOL> os . rename ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT>' . format ( self . get_head ( ) ) ) , <EOL> os . path . join ( self . tmppath , '<STR_LIT>' , new_name ) ) <EOL> def list_objects ( self ) : <EOL> return set ( int ( key ) for key in self . repository . list ( ) ) <EOL> def test_repair_corrupted_segment ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ] ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ) , self . list_objects ( ) ) <EOL> self . check ( status = True ) <EOL> self . corrupt_object ( <NUM_LIT:5> ) <EOL> self . assert_raises ( IntegrityError , lambda : self . get_objects ( <NUM_LIT:5> ) ) <EOL> self . repository . rollback ( ) <EOL> self . check ( status = False ) <EOL> self . check ( status = False ) <EOL> self . check ( repair = True , status = True ) <EOL> self . get_objects ( <NUM_LIT:4> ) <EOL> self . check ( status = True ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:6> ] ) , self . list_objects ( ) ) <EOL> def test_repair_missing_segment ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ] ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ) , self . list_objects ( ) ) <EOL> self . check ( status = True ) <EOL> self . delete_segment ( <NUM_LIT:1> ) <EOL> self . repository . rollback ( ) <EOL> self . check ( repair = True , status = True ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , self . list_objects ( ) ) <EOL> def test_repair_missing_commit_segment ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ] ) <EOL> self . delete_segment ( <NUM_LIT:1> ) <EOL> self . assert_raises ( Repository . ObjectNotFound , lambda : self . get_objects ( <NUM_LIT:4> ) ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , self . list_objects ( ) ) <EOL> def test_repair_corrupted_commit_segment ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ] ) <EOL> with open ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT:data>' , '<STR_LIT:0>' , '<STR_LIT:1>' ) , '<STR_LIT>' ) as fd : <EOL> fd . seek ( - <NUM_LIT:1> , os . SEEK_END ) <EOL> fd . write ( b'<STR_LIT:X>' ) <EOL> self . assert_raises ( Repository . ObjectNotFound , lambda : self . get_objects ( <NUM_LIT:4> ) ) <EOL> self . check ( status = True ) <EOL> self . get_objects ( <NUM_LIT:3> ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , self . list_objects ( ) ) <EOL> def test_repair_no_commits ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ] ) <EOL> with open ( os . path . join ( self . tmppath , '<STR_LIT>' , '<STR_LIT:data>' , '<STR_LIT:0>' , '<STR_LIT:0>' ) , '<STR_LIT>' ) as fd : <EOL> fd . seek ( - <NUM_LIT:1> , os . SEEK_END ) <EOL> fd . write ( b'<STR_LIT:X>' ) <EOL> self . assert_raises ( Repository . CheckNeeded , lambda : self . get_objects ( <NUM_LIT:4> ) ) <EOL> self . check ( status = False ) <EOL> self . check ( status = False ) <EOL> self . assert_equal ( self . list_indices ( ) , [ '<STR_LIT>' ] ) <EOL> self . check ( repair = True , status = True ) <EOL> self . assert_equal ( self . list_indices ( ) , [ '<STR_LIT>' ] ) <EOL> self . check ( status = True ) <EOL> self . get_objects ( <NUM_LIT:3> ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , self . list_objects ( ) ) <EOL> def test_repair_missing_index ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ] ) <EOL> self . delete_index ( ) <EOL> self . check ( status = True ) <EOL> self . get_objects ( <NUM_LIT:4> ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ) , self . list_objects ( ) ) <EOL> def test_repair_index_too_new ( self ) : <EOL> self . add_objects ( [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ] ) <EOL> self . assert_equal ( self . list_indices ( ) , [ '<STR_LIT>' ] ) <EOL> self . rename_index ( '<STR_LIT>' ) <EOL> self . check ( status = True ) <EOL> self . assert_equal ( self . list_indices ( ) , [ '<STR_LIT>' ] ) <EOL> self . get_objects ( <NUM_LIT:4> ) <EOL> self . assert_equal ( set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ) , self . list_objects ( ) ) <EOL> def test_crash_before_compact ( self ) : <EOL> self . repository . put ( bytes ( <NUM_LIT:32> ) , b'<STR_LIT:data>' ) <EOL> self . repository . put ( bytes ( <NUM_LIT:32> ) , b'<STR_LIT>' ) <EOL> with patch . object ( Repository , '<STR_LIT>' ) as compact : <EOL> self . repository . commit ( ) <EOL> compact . assert_called_once_with ( save_space = False ) <EOL> self . reopen ( ) <EOL> with self . repository : <EOL> self . check ( repair = True ) <EOL> self . assert_equal ( self . repository . get ( bytes ( <NUM_LIT:32> ) ) , b'<STR_LIT>' ) <EOL> class RemoteRepositoryTestCase ( RepositoryTestCase ) : <EOL> def open ( self , create = False ) : <EOL> return RemoteRepository ( Location ( '<STR_LIT>' + os . path . join ( self . tmppath , '<STR_LIT>' ) ) , create = create ) <EOL> def test_invalid_rpc ( self ) : <EOL> self . assert_raises ( InvalidRPCMethod , lambda : self . repository . call ( '<STR_LIT>' , None ) ) <EOL> def test_ssh_cmd ( self ) : <EOL> assert self . repository . ssh_cmd ( Location ( '<STR_LIT>' ) ) == [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> assert self . repository . ssh_cmd ( Location ( '<STR_LIT>' ) ) == [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> assert self . repository . ssh_cmd ( Location ( '<STR_LIT>' ) ) == [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> assert self . repository . ssh_cmd ( Location ( '<STR_LIT>' ) ) == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> assert self . repository . ssh_cmd ( Location ( '<STR_LIT>' ) ) == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def test_borg_cmd ( self ) : <EOL> class MockArgs : <EOL> remote_path = '<STR_LIT>' <EOL> umask = <NUM_LIT> <EOL> assert self . repository . borg_cmd ( None , testing = True ) == [ sys . executable , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> args = MockArgs ( ) <EOL> assert self . repository . borg_cmd ( args , testing = False ) == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> args . remote_path = '<STR_LIT>' <EOL> assert self . repository . borg_cmd ( args , testing = False ) == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class RemoteRepositoryCheckTestCase ( RepositoryCheckTestCase ) : <EOL> def open ( self , create = False ) : <EOL> return RemoteRepository ( Location ( '<STR_LIT>' + os . path . join ( self . tmppath , '<STR_LIT>' ) ) , create = create ) <EOL> def test_crash_before_compact ( self ) : <EOL> pass </s>
