<s> from collections import defaultdict <EOL> import mock <EOL> import os <EOL> import time <EOL> import unittest <EOL> import eventlet <EOL> import itertools <EOL> from six . moves import urllib <EOL> from swift . common . exceptions import DiskFileNotExist , DiskFileError , DiskFileDeleted <EOL> from swift . common import utils <EOL> from swift . common . storage_policy import POLICIES <EOL> from swift . common . utils import Timestamp <EOL> from swift . obj import ssync_sender , server <EOL> from swift . obj . reconstructor import RebuildingECDiskFileStream <EOL> from test . unit import patch_policies <EOL> from test . unit . obj . common import BaseTest , FakeReplicator <EOL> class TestBaseSsync ( BaseTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestBaseSsync , self ) . setUp ( ) <EOL> self . device = '<STR_LIT>' <EOL> self . partition = '<STR_LIT>' <EOL> self . tx_testdir = os . path . join ( self . tmpdir , '<STR_LIT>' ) <EOL> utils . mkdirs ( os . path . join ( self . tx_testdir , self . device ) ) <EOL> self . daemon = FakeReplicator ( self . tx_testdir ) <EOL> self . rx_testdir = os . path . join ( self . tmpdir , '<STR_LIT>' ) <EOL> utils . mkdirs ( os . path . join ( self . rx_testdir , self . device ) ) <EOL> conf = { <EOL> '<STR_LIT>' : self . rx_testdir , <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT:false>' } <EOL> self . rx_controller = server . ObjectController ( conf ) <EOL> self . ts_iter = ( Timestamp ( t ) <EOL> for t in itertools . count ( int ( time . time ( ) ) ) ) <EOL> self . rx_ip = '<STR_LIT:127.0.0.1>' <EOL> sock = eventlet . listen ( ( self . rx_ip , <NUM_LIT:0> ) ) <EOL> self . rx_server = eventlet . spawn ( <EOL> eventlet . wsgi . server , sock , self . rx_controller , utils . NullLogger ( ) ) <EOL> self . rx_port = sock . getsockname ( ) [ <NUM_LIT:1> ] <EOL> self . rx_node = { '<STR_LIT>' : self . rx_ip , <EOL> '<STR_LIT>' : self . rx_port , <EOL> '<STR_LIT>' : self . device } <EOL> def tearDown ( self ) : <EOL> self . rx_server . kill ( ) <EOL> super ( TestBaseSsync , self ) . tearDown ( ) <EOL> def make_connect_wrapper ( self , sender ) : <EOL> """<STR_LIT>""" <EOL> orig_connect = sender . connect <EOL> trace = dict ( messages = [ ] ) <EOL> def add_trace ( type , msg ) : <EOL> if msg . strip ( ) : <EOL> trace [ '<STR_LIT>' ] . append ( ( type , msg . strip ( ) ) ) <EOL> def make_send_wrapper ( send ) : <EOL> def wrapped_send ( msg ) : <EOL> _msg = msg . split ( '<STR_LIT:\r\n>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> _msg = _msg . rsplit ( '<STR_LIT:\r\n>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> add_trace ( '<STR_LIT>' , _msg ) <EOL> send ( msg ) <EOL> return wrapped_send <EOL> def make_readline_wrapper ( readline ) : <EOL> def wrapped_readline ( ) : <EOL> data = readline ( ) <EOL> add_trace ( '<STR_LIT>' , data ) <EOL> bytes_read = trace . setdefault ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> trace [ '<STR_LIT>' ] = bytes_read + len ( data ) <EOL> return data <EOL> return wrapped_readline <EOL> def wrapped_connect ( ) : <EOL> orig_connect ( ) <EOL> sender . connection . send = make_send_wrapper ( <EOL> sender . connection . send ) <EOL> sender . readline = make_readline_wrapper ( sender . readline ) <EOL> return wrapped_connect , trace <EOL> def _create_ondisk_files ( self , df_mgr , obj_name , policy , timestamp , <EOL> frag_indexes = None , commit = True ) : <EOL> frag_indexes = [ None ] if frag_indexes is None else frag_indexes <EOL> metadata = { '<STR_LIT:Content-Type>' : '<STR_LIT>' } <EOL> diskfiles = [ ] <EOL> for frag_index in frag_indexes : <EOL> object_data = '<STR_LIT>' % ( obj_name , frag_index ) <EOL> if frag_index is not None : <EOL> metadata [ '<STR_LIT>' ] = str ( frag_index ) <EOL> df = self . _make_diskfile ( <EOL> device = self . device , partition = self . partition , account = '<STR_LIT:a>' , <EOL> container = '<STR_LIT:c>' , obj = obj_name , body = object_data , <EOL> extra_metadata = metadata , timestamp = timestamp , policy = policy , <EOL> frag_index = frag_index , df_mgr = df_mgr , commit = commit ) <EOL> if commit : <EOL> df . open ( ) <EOL> listing = os . listdir ( df . _datadir ) <EOL> self . assertTrue ( listing ) <EOL> for filename in listing : <EOL> self . assertTrue ( filename . startswith ( timestamp . internal ) ) <EOL> diskfiles . append ( df ) <EOL> return diskfiles <EOL> def _open_tx_diskfile ( self , obj_name , policy , frag_index = None ) : <EOL> df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> df = df_mgr . get_diskfile ( <EOL> self . device , self . partition , account = '<STR_LIT:a>' , container = '<STR_LIT:c>' , <EOL> obj = obj_name , policy = policy , frag_index = frag_index ) <EOL> df . open ( ) <EOL> return df <EOL> def _open_rx_diskfile ( self , obj_name , policy , frag_index = None ) : <EOL> df = self . rx_controller . get_diskfile ( <EOL> self . device , self . partition , '<STR_LIT:a>' , '<STR_LIT:c>' , obj_name , policy = policy , <EOL> frag_index = frag_index ) <EOL> df . open ( ) <EOL> return df <EOL> def _verify_diskfile_sync ( self , tx_df , rx_df , frag_index , same_etag = False ) : <EOL> self . assertNotEqual ( tx_df . _datadir , rx_df . _datadir ) <EOL> rx_metadata = dict ( rx_df . get_metadata ( ) ) <EOL> for k , v in tx_df . get_metadata ( ) . items ( ) : <EOL> if k == '<STR_LIT>' : <EOL> self . assertTrue ( k in rx_metadata ) <EOL> self . assertEqual ( frag_index , int ( rx_metadata . pop ( k ) ) ) <EOL> elif k == '<STR_LIT>' and not same_etag : <EOL> self . assertNotEqual ( v , rx_metadata . pop ( k , None ) ) <EOL> continue <EOL> else : <EOL> self . assertEqual ( v , rx_metadata . pop ( k ) , k ) <EOL> self . assertFalse ( rx_metadata ) <EOL> expected_body = '<STR_LIT>' % ( tx_df . _name , frag_index ) <EOL> actual_body = '<STR_LIT>' . join ( [ chunk for chunk in rx_df . reader ( ) ] ) <EOL> self . assertEqual ( expected_body , actual_body ) <EOL> def _analyze_trace ( self , trace ) : <EOL> """<STR_LIT>""" <EOL> def tx_missing ( results , line ) : <EOL> self . assertEqual ( '<STR_LIT>' , line [ <NUM_LIT:0> ] ) <EOL> results [ '<STR_LIT>' ] . append ( line [ <NUM_LIT:1> ] ) <EOL> def rx_missing ( results , line ) : <EOL> self . assertEqual ( '<STR_LIT>' , line [ <NUM_LIT:0> ] ) <EOL> parts = line [ <NUM_LIT:1> ] . split ( '<STR_LIT:\r\n>' ) <EOL> for part in parts : <EOL> results [ '<STR_LIT>' ] . append ( part ) <EOL> def tx_updates ( results , line ) : <EOL> self . assertEqual ( '<STR_LIT>' , line [ <NUM_LIT:0> ] ) <EOL> subrequests = results [ '<STR_LIT>' ] <EOL> if line [ <NUM_LIT:1> ] . startswith ( ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:POST>' ) ) : <EOL> parts = line [ <NUM_LIT:1> ] . split ( '<STR_LIT:\r\n>' ) <EOL> method , path = parts [ <NUM_LIT:0> ] . split ( ) <EOL> subreq = { '<STR_LIT>' : method , '<STR_LIT:path>' : path , '<STR_LIT>' : line [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : parts [ <NUM_LIT:1> : ] } <EOL> subrequests . append ( subreq ) <EOL> else : <EOL> self . assertTrue ( subrequests ) <EOL> body = ( subrequests [ - <NUM_LIT:1> ] ) . setdefault ( '<STR_LIT:body>' , '<STR_LIT>' ) <EOL> body += line [ <NUM_LIT:1> ] <EOL> subrequests [ - <NUM_LIT:1> ] [ '<STR_LIT:body>' ] = body <EOL> def rx_updates ( results , line ) : <EOL> self . assertEqual ( '<STR_LIT>' , line [ <NUM_LIT:0> ] ) <EOL> results . setdefault [ '<STR_LIT>' ] . append ( line [ <NUM_LIT:1> ] ) <EOL> def unexpected ( results , line ) : <EOL> results . setdefault ( '<STR_LIT>' , [ ] ) . append ( line ) <EOL> handshakes = iter ( [ ( ( '<STR_LIT>' , '<STR_LIT>' ) , tx_missing ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , unexpected ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , rx_missing ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , unexpected ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , tx_updates ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , unexpected ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , rx_updates ) , <EOL> ( ( '<STR_LIT>' , '<STR_LIT>' ) , unexpected ) ] ) <EOL> expect_handshake = next ( handshakes ) <EOL> phases = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> results = dict ( ( k , [ ] ) for k in phases ) <EOL> handler = unexpected <EOL> lines = list ( trace . get ( '<STR_LIT>' , [ ] ) ) <EOL> lines . reverse ( ) <EOL> while lines : <EOL> line = lines . pop ( ) <EOL> if line == expect_handshake [ <NUM_LIT:0> ] : <EOL> handler = expect_handshake [ <NUM_LIT:1> ] <EOL> try : <EOL> expect_handshake = next ( handshakes ) <EOL> except StopIteration : <EOL> self . assertFalse ( <EOL> lines , '<STR_LIT>' % lines ) <EOL> continue <EOL> handler ( results , line ) <EOL> try : <EOL> missed = next ( handshakes ) <EOL> self . fail ( '<STR_LIT>' % str ( missed [ <NUM_LIT:0> ] ) ) <EOL> except StopIteration : <EOL> pass <EOL> self . assertFalse ( results . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' % results . get ( None ) ) <EOL> return results <EOL> def _verify_ondisk_files ( self , tx_objs , policy , tx_frag_index = None , <EOL> rx_frag_index = None ) : <EOL> """<STR_LIT>""" <EOL> for o_name , diskfiles in tx_objs . items ( ) : <EOL> for tx_df in diskfiles : <EOL> tx_df . open ( ) <EOL> if tx_frag_index is None or tx_df . _frag_index == tx_frag_index : <EOL> rx_df = self . _open_rx_diskfile ( <EOL> o_name , policy , rx_frag_index ) <EOL> match_etag = ( tx_frag_index == rx_frag_index ) <EOL> self . _verify_diskfile_sync ( <EOL> tx_df , rx_df , rx_frag_index , match_etag ) <EOL> else : <EOL> self . assertRaises ( DiskFileNotExist , self . _open_rx_diskfile , <EOL> o_name , policy , <EOL> frag_index = tx_df . _frag_index ) <EOL> def _verify_tombstones ( self , tx_objs , policy ) : <EOL> for o_name , diskfiles in tx_objs . items ( ) : <EOL> try : <EOL> self . _open_tx_diskfile ( o_name , policy ) <EOL> self . fail ( '<STR_LIT>' ) <EOL> except DiskFileDeleted as exc : <EOL> tx_delete_time = exc . timestamp <EOL> try : <EOL> self . _open_rx_diskfile ( o_name , policy ) <EOL> self . fail ( '<STR_LIT>' ) <EOL> except DiskFileDeleted as exc : <EOL> rx_delete_time = exc . timestamp <EOL> self . assertEqual ( tx_delete_time , rx_delete_time ) <EOL> @ patch_policies ( with_ec_default = True ) <EOL> class TestSsyncEC ( TestBaseSsync ) : <EOL> def test_handoff_fragment_revert ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = <NUM_LIT:0> <EOL> tx_node_index = <NUM_LIT:1> <EOL> frag_index = rx_node_index <EOL> tx_objs = { } <EOL> rx_objs = { } <EOL> tx_tombstones = { } <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> t1 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t1 , ( rx_node_index , tx_node_index ) ) <EOL> t2 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t2 , ( tx_node_index , ) ) <EOL> t3 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t3 , ( rx_node_index , ) ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT>' , policy , t3 , ( <NUM_LIT> , ) ) <EOL> t4 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t4 , ( tx_node_index , rx_node_index , ) ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT>' , policy , t4 , ( rx_node_index , ) ) <EOL> t5 = next ( self . ts_iter ) <EOL> tx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t5 , ( tx_node_index , ) ) <EOL> tx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( t5 ) <EOL> suffixes = set ( ) <EOL> for diskfiles in ( tx_objs . values ( ) + tx_tombstones . values ( ) ) : <EOL> for df in diskfiles : <EOL> suffixes . add ( os . path . basename ( os . path . dirname ( df . _datadir ) ) ) <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy , <EOL> '<STR_LIT>' : frag_index } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> sender ( ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:4> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> sync_paths = [ ] <EOL> for subreq in results . get ( '<STR_LIT>' ) : <EOL> if subreq . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . assertTrue ( <EOL> '<STR_LIT>' % rx_node_index <EOL> in subreq . get ( '<STR_LIT>' ) ) <EOL> expected_body = '<STR_LIT>' % ( subreq [ '<STR_LIT:path>' ] , rx_node_index ) <EOL> self . assertEqual ( expected_body , subreq [ '<STR_LIT:body>' ] ) <EOL> elif subreq . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . assertEqual ( '<STR_LIT>' , subreq [ '<STR_LIT:path>' ] ) <EOL> sync_paths . append ( subreq . get ( '<STR_LIT:path>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , sorted ( sync_paths ) ) <EOL> self . _verify_ondisk_files ( <EOL> tx_objs , policy , frag_index , rx_node_index ) <EOL> self . _verify_tombstones ( tx_tombstones , policy ) <EOL> def test_handoff_fragment_only_missing_durable ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = frag_index = <NUM_LIT:0> <EOL> tx_node_index = <NUM_LIT:1> <EOL> tx_objs = { } <EOL> rx_objs = { } <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> expected_subreqs = defaultdict ( list ) <EOL> t1a = next ( self . ts_iter ) <EOL> t1b = next ( self . ts_iter ) <EOL> t1c = next ( self . ts_iter ) <EOL> obj_name = '<STR_LIT>' <EOL> tx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , obj_name , policy , t1c , ( tx_node_index , rx_node_index , ) ) <EOL> rx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , obj_name , policy , t1a , ( rx_node_index , ) ) <EOL> metadata = { '<STR_LIT>' : t1b . internal } <EOL> rx_objs [ obj_name ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> rx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , obj_name , policy , t1c , ( rx_node_index , <NUM_LIT:9> ) , commit = False ) <EOL> t2 = next ( self . ts_iter ) <EOL> obj_name = '<STR_LIT>' <EOL> tx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , obj_name , policy , t2 , ( tx_node_index , rx_node_index , ) ) <EOL> rx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , obj_name , policy , t2 , ( <NUM_LIT> , <NUM_LIT> ) , commit = False ) <EOL> expected_subreqs [ '<STR_LIT>' ] . append ( obj_name ) <EOL> t3 = next ( self . ts_iter ) <EOL> obj_name = '<STR_LIT>' <EOL> tx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , obj_name , policy , t3 , ( tx_node_index , rx_node_index , ) ) <EOL> t3b = next ( self . ts_iter ) <EOL> rx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , obj_name , policy , t3b , ( rx_node_index , ) , commit = False ) <EOL> expected_subreqs [ '<STR_LIT>' ] . append ( obj_name ) <EOL> t4 = next ( self . ts_iter ) <EOL> obj_name = '<STR_LIT>' <EOL> tx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , obj_name , policy , t4 , ( tx_node_index , rx_node_index , ) ) <EOL> rx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , obj_name , policy , t4 , ( rx_node_index , ) ) <EOL> t4b = next ( self . ts_iter ) <EOL> rx_objs [ obj_name ] [ <NUM_LIT:0> ] . delete ( t4b ) <EOL> t4c = next ( self . ts_iter ) <EOL> rx_objs [ obj_name ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , obj_name , policy , t4c , ( rx_node_index , ) , commit = False ) <EOL> suffixes = set ( ) <EOL> for diskfiles in tx_objs . values ( ) : <EOL> for df in diskfiles : <EOL> suffixes . add ( os . path . basename ( os . path . dirname ( df . _datadir ) ) ) <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy , <EOL> '<STR_LIT>' : frag_index } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> sender ( ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:4> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> for subreq in results . get ( '<STR_LIT>' ) : <EOL> obj = subreq [ '<STR_LIT:path>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:3> ] <EOL> method = subreq [ '<STR_LIT>' ] <EOL> self . assertTrue ( obj in expected_subreqs [ method ] , <EOL> '<STR_LIT>' <EOL> % ( method , obj , expected_subreqs [ method ] ) ) <EOL> expected_subreqs [ method ] . remove ( obj ) <EOL> if method == '<STR_LIT>' : <EOL> expected_body = '<STR_LIT>' % ( subreq [ '<STR_LIT:path>' ] , rx_node_index ) <EOL> self . assertEqual ( expected_body , subreq [ '<STR_LIT:body>' ] ) <EOL> for _method , expected in expected_subreqs . items ( ) : <EOL> self . assertFalse ( expected ) <EOL> tx_objs . pop ( '<STR_LIT>' ) <EOL> self . _verify_ondisk_files ( <EOL> tx_objs , policy , frag_index , rx_node_index ) <EOL> def test_fragment_sync ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = <NUM_LIT:0> <EOL> tx_node_index = <NUM_LIT:1> <EOL> frag_index = tx_node_index <EOL> tx_objs = { } <EOL> tx_tombstones = { } <EOL> rx_objs = { } <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> t1 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t1 , ( tx_node_index , ) ) <EOL> t2 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t2 , ( tx_node_index , ) ) <EOL> t3 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t3 , ( tx_node_index , ) ) <EOL> t4 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t4 , ( tx_node_index , ) ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT>' , policy , t4 , ( rx_node_index , ) ) <EOL> t5 = next ( self . ts_iter ) <EOL> tx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t5 , ( tx_node_index , ) ) <EOL> tx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( t5 ) <EOL> suffixes = set ( ) <EOL> for diskfiles in ( tx_objs . values ( ) + tx_tombstones . values ( ) ) : <EOL> for df in diskfiles : <EOL> suffixes . add ( os . path . basename ( os . path . dirname ( df . _datadir ) ) ) <EOL> reconstruct_fa_calls = [ ] <EOL> def fake_reconstruct_fa ( job , node , metadata ) : <EOL> reconstruct_fa_calls . append ( ( job , node , policy , metadata ) ) <EOL> if len ( reconstruct_fa_calls ) == <NUM_LIT:2> : <EOL> raise DiskFileError <EOL> content = '<STR_LIT>' % ( metadata [ '<STR_LIT:name>' ] , rx_node_index ) <EOL> return RebuildingECDiskFileStream ( <EOL> metadata , rx_node_index , iter ( [ content ] ) ) <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy , <EOL> '<STR_LIT>' : frag_index , <EOL> '<STR_LIT>' : fake_reconstruct_fa } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> sender ( ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:4> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( reconstruct_fa_calls ) ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> actual_sync_paths = [ ] <EOL> for subreq in results . get ( '<STR_LIT>' ) : <EOL> if subreq . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . assertTrue ( <EOL> '<STR_LIT>' % rx_node_index <EOL> in subreq . get ( '<STR_LIT>' ) ) <EOL> expected_body = '<STR_LIT>' % ( subreq [ '<STR_LIT:path>' ] , rx_node_index ) <EOL> self . assertEqual ( expected_body , subreq [ '<STR_LIT:body>' ] ) <EOL> elif subreq . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . assertEqual ( '<STR_LIT>' , subreq [ '<STR_LIT:path>' ] ) <EOL> actual_sync_paths . append ( subreq . get ( '<STR_LIT:path>' ) ) <EOL> expect_sync_paths = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> failed_path = reconstruct_fa_calls [ <NUM_LIT:1> ] [ <NUM_LIT:3> ] [ '<STR_LIT:name>' ] <EOL> expect_sync_paths . remove ( failed_path ) <EOL> failed_obj = None <EOL> for obj , diskfiles in tx_objs . items ( ) : <EOL> if diskfiles [ <NUM_LIT:0> ] . _name == failed_path : <EOL> failed_obj = obj <EOL> self . assertTrue ( tx_objs . pop ( failed_obj ) ) <EOL> self . assertEqual ( sorted ( expect_sync_paths ) , sorted ( actual_sync_paths ) ) <EOL> self . _verify_ondisk_files ( <EOL> tx_objs , policy , frag_index , rx_node_index ) <EOL> self . _verify_tombstones ( tx_tombstones , policy ) <EOL> def test_send_with_frag_index_none ( self ) : <EOL> policy = POLICIES . default <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> ts1 = next ( self . ts_iter ) <EOL> remote_df = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT:o>' , policy , ts1 , ( <NUM_LIT:3> , ) ) [ <NUM_LIT:0> ] <EOL> df = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT:o>' , policy , ts1 , ( <NUM_LIT:3> , ) ) [ <NUM_LIT:0> ] <EOL> suffix = os . path . basename ( os . path . dirname ( df . _datadir ) ) <EOL> ts2 = next ( self . ts_iter ) <EOL> df . delete ( ts2 ) <EOL> job = { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy , <EOL> '<STR_LIT>' : self . device , <EOL> } <EOL> sender = ssync_sender . Sender ( <EOL> self . daemon , self . rx_node , job , [ suffix ] ) <EOL> success , _ = sender ( ) <EOL> self . assertTrue ( success ) <EOL> try : <EOL> remote_df . read_metadata ( ) <EOL> except DiskFileDeleted as e : <EOL> self . assertEqual ( e . timestamp , ts2 ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_send_invalid_frag_index ( self ) : <EOL> policy = POLICIES . default <EOL> job = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy } <EOL> sender = ssync_sender . Sender ( <EOL> self . daemon , self . rx_node , job , [ '<STR_LIT:abc>' ] ) <EOL> success , _ = sender ( ) <EOL> self . assertFalse ( success ) <EOL> error_log_lines = self . daemon . logger . get_lines_for_level ( '<STR_LIT:error>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( error_log_lines ) ) <EOL> error_msg = error_log_lines [ <NUM_LIT:0> ] <EOL> self . assertIn ( "<STR_LIT>" , error_msg ) <EOL> self . assertIn ( "<STR_LIT>" , <EOL> error_msg ) <EOL> @ patch_policies <EOL> class TestSsyncReplication ( TestBaseSsync ) : <EOL> def test_sync ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = <NUM_LIT:0> <EOL> tx_objs = { } <EOL> rx_objs = { } <EOL> tx_tombstones = { } <EOL> rx_tombstones = { } <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> t1 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t1 ) <EOL> t2 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t2 ) <EOL> t3a = next ( self . ts_iter ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t3a ) <EOL> t3b = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t3b ) <EOL> t4 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t4 ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t4 ) <EOL> t5 = next ( self . ts_iter ) <EOL> tx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t5 ) <EOL> tx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( t5 ) <EOL> t6 = next ( self . ts_iter ) <EOL> tx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t6 ) <EOL> tx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( t6 ) <EOL> rx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT>' , policy , t6 ) <EOL> rx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( t6 ) <EOL> t7a = next ( self . ts_iter ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t7a ) <EOL> t7b = next ( self . ts_iter ) <EOL> tx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t7b ) <EOL> tx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( t7b ) <EOL> suffixes = set ( ) <EOL> for diskfiles in ( tx_objs . values ( ) + tx_tombstones . values ( ) ) : <EOL> for df in diskfiles : <EOL> suffixes . add ( os . path . basename ( os . path . dirname ( df . _datadir ) ) ) <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> success , in_sync_objs = sender ( ) <EOL> self . assertEqual ( <NUM_LIT:7> , len ( in_sync_objs ) ) <EOL> self . assertTrue ( success ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:7> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> sync_paths = [ ] <EOL> for subreq in results . get ( '<STR_LIT>' ) : <EOL> if subreq . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . assertTrue ( <EOL> subreq [ '<STR_LIT:path>' ] in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> expected_body = '<STR_LIT>' % subreq [ '<STR_LIT:path>' ] <EOL> self . assertEqual ( expected_body , subreq [ '<STR_LIT:body>' ] ) <EOL> elif subreq . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> self . assertTrue ( subreq [ '<STR_LIT:path>' ] in ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> sync_paths . append ( subreq . get ( '<STR_LIT:path>' ) ) <EOL> self . assertEqual ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> sorted ( sync_paths ) ) <EOL> self . _verify_ondisk_files ( tx_objs , policy ) <EOL> self . _verify_tombstones ( tx_tombstones , policy ) <EOL> def test_nothing_to_sync ( self ) : <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : POLICIES . default } <EOL> node = { '<STR_LIT>' : self . rx_ip , <EOL> '<STR_LIT>' : self . rx_port , <EOL> '<STR_LIT>' : self . device , <EOL> '<STR_LIT:index>' : <NUM_LIT:0> } <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , [ '<STR_LIT:abc>' ] ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> result , in_sync_objs = sender ( ) <EOL> self . assertTrue ( result ) <EOL> self . assertFalse ( in_sync_objs ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT> , trace . get ( '<STR_LIT>' ) ) <EOL> def test_meta_file_sync ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = <NUM_LIT:0> <EOL> tx_objs = { } <EOL> rx_objs = { } <EOL> tx_tombstones = { } <EOL> rx_tombstones = { } <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> expected_subreqs = defaultdict ( list ) <EOL> t1 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t1 ) <EOL> t1_meta = next ( self . ts_iter ) <EOL> metadata = { '<STR_LIT>' : t1_meta . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> expected_subreqs [ '<STR_LIT>' ] . append ( '<STR_LIT>' ) <EOL> expected_subreqs [ '<STR_LIT:POST>' ] . append ( '<STR_LIT>' ) <EOL> t2 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t2 ) <EOL> t2_meta = next ( self . ts_iter ) <EOL> metadata = { '<STR_LIT>' : t2_meta . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t2 ) <EOL> expected_subreqs [ '<STR_LIT:POST>' ] . append ( '<STR_LIT>' ) <EOL> t3a = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t3a ) <EOL> t3b = next ( self . ts_iter ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t3b ) <EOL> t3_meta = next ( self . ts_iter ) <EOL> metadata = { '<STR_LIT>' : t3_meta . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> expected_subreqs [ '<STR_LIT:POST>' ] . append ( '<STR_LIT>' ) <EOL> t4a = next ( self . ts_iter ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t4a ) <EOL> t4b = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t4b ) <EOL> t4_meta = next ( self . ts_iter ) <EOL> metadata = { '<STR_LIT>' : t4_meta . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> rx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> expected_subreqs [ '<STR_LIT>' ] . append ( '<STR_LIT>' ) <EOL> t5 = next ( self . ts_iter ) <EOL> rx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t5 ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t5 ) <EOL> t5_meta = next ( self . ts_iter ) <EOL> metadata = { '<STR_LIT>' : t5_meta . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> rx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> t6 = next ( self . ts_iter ) <EOL> tx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> tx_df_mgr , '<STR_LIT>' , policy , t6 ) <EOL> rx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT>' , policy , t6 ) <EOL> metadata = { '<STR_LIT>' : next ( self . ts_iter ) . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> tx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( next ( self . ts_iter ) ) <EOL> expected_subreqs [ '<STR_LIT>' ] . append ( '<STR_LIT>' ) <EOL> t7 = next ( self . ts_iter ) <EOL> tx_objs [ '<STR_LIT>' ] = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t7 ) <EOL> rx_tombstones [ '<STR_LIT>' ] = self . _create_ondisk_files ( <EOL> rx_df_mgr , '<STR_LIT>' , policy , t7 ) <EOL> metadata = { '<STR_LIT>' : next ( self . ts_iter ) . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_objs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> rx_tombstones [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . delete ( next ( self . ts_iter ) ) <EOL> suffixes = set ( ) <EOL> for diskfiles in ( tx_objs . values ( ) + tx_tombstones . values ( ) ) : <EOL> for df in diskfiles : <EOL> suffixes . add ( os . path . basename ( os . path . dirname ( df . _datadir ) ) ) <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> success , in_sync_objs = sender ( ) <EOL> self . assertEqual ( <NUM_LIT:7> , len ( in_sync_objs ) ) <EOL> self . assertTrue ( success ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:7> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( results [ '<STR_LIT>' ] ) ) <EOL> for subreq in results . get ( '<STR_LIT>' ) : <EOL> obj = subreq [ '<STR_LIT:path>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:3> ] <EOL> method = subreq [ '<STR_LIT>' ] <EOL> self . assertTrue ( obj in expected_subreqs [ method ] , <EOL> '<STR_LIT>' <EOL> % ( method , obj , expected_subreqs [ method ] ) ) <EOL> expected_subreqs [ method ] . remove ( obj ) <EOL> if method == '<STR_LIT>' : <EOL> expected_body = '<STR_LIT>' % subreq [ '<STR_LIT:path>' ] <EOL> self . assertEqual ( expected_body , subreq [ '<STR_LIT:body>' ] ) <EOL> for _method , expected in expected_subreqs . items ( ) : <EOL> self . assertFalse ( expected ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> del tx_objs [ '<STR_LIT>' ] <EOL> self . _verify_ondisk_files ( tx_objs , policy ) <EOL> self . _verify_tombstones ( tx_tombstones , policy ) <EOL> for oname , rx_obj in rx_objs . items ( ) : <EOL> df = rx_obj [ <NUM_LIT:0> ] . open ( ) <EOL> metadata = df . get_metadata ( ) <EOL> self . assertEqual ( metadata [ '<STR_LIT>' ] , oname ) <EOL> self . assertEqual ( metadata [ '<STR_LIT>' ] , '<STR_LIT>' + oname ) <EOL> def test_meta_file_not_synced_to_legacy_receiver ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = <NUM_LIT:0> <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> t1 = next ( self . ts_iter ) <EOL> self . _create_ondisk_files ( rx_df_mgr , '<STR_LIT>' , policy , t1 ) <EOL> t2 = next ( self . ts_iter ) <EOL> tx_obj = self . _create_ondisk_files ( tx_df_mgr , '<STR_LIT>' , policy , t2 ) [ <NUM_LIT:0> ] <EOL> t3 = next ( self . ts_iter ) <EOL> metadata = { '<STR_LIT>' : t3 . internal , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> tx_obj . write_metadata ( metadata ) <EOL> suffixes = [ os . path . basename ( os . path . dirname ( tx_obj . _datadir ) ) ] <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> def _legacy_check_missing ( self , line ) : <EOL> parts = line . split ( ) <EOL> object_hash = urllib . parse . unquote ( parts [ <NUM_LIT:0> ] ) <EOL> timestamp = urllib . parse . unquote ( parts [ <NUM_LIT:1> ] ) <EOL> want = False <EOL> try : <EOL> df = self . diskfile_mgr . get_diskfile_from_hash ( <EOL> self . device , self . partition , object_hash , self . policy , <EOL> frag_index = self . frag_index ) <EOL> except DiskFileNotExist : <EOL> want = True <EOL> else : <EOL> try : <EOL> df . open ( ) <EOL> except DiskFileDeleted as err : <EOL> want = err . timestamp < timestamp <EOL> except DiskFileError : <EOL> want = True <EOL> else : <EOL> want = df . timestamp < timestamp <EOL> if want : <EOL> return urllib . parse . quote ( object_hash ) <EOL> return None <EOL> func = '<STR_LIT>' <EOL> with mock . patch ( func , _legacy_check_missing ) : <EOL> success , in_sync_objs = sender ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( in_sync_objs ) ) <EOL> self . assertTrue ( success ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( '<STR_LIT>' , results [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> rx_obj = self . _open_rx_diskfile ( '<STR_LIT>' , policy ) <EOL> tx_obj = self . _open_tx_diskfile ( '<STR_LIT>' , policy ) <EOL> self . assertEqual ( t2 , rx_obj . data_timestamp ) <EOL> self . assertEqual ( t2 , rx_obj . timestamp ) <EOL> self . assertEqual ( rx_obj . data_timestamp , tx_obj . data_timestamp ) <EOL> self . assertNotIn ( '<STR_LIT>' , rx_obj . get_metadata ( ) ) <EOL> def test_content_type_sync ( self ) : <EOL> policy = POLICIES . default <EOL> rx_node_index = <NUM_LIT:0> <EOL> tx_objs = { } <EOL> rx_objs = { } <EOL> tx_df_mgr = self . daemon . _diskfile_router [ policy ] <EOL> rx_df_mgr = self . rx_controller . _diskfile_router [ policy ] <EOL> expected_subreqs = defaultdict ( list ) <EOL> name = '<STR_LIT>' <EOL> t1 = self . ts_iter . next ( ) <EOL> tx_objs [ name ] = self . _create_ondisk_files ( tx_df_mgr , name , policy , t1 ) <EOL> t1_type = self . ts_iter . next ( ) <EOL> metadata_1 = { '<STR_LIT>' : t1_type . internal , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t1_type . internal } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_1 ) <EOL> t1_meta = self . ts_iter . next ( ) <EOL> metadata_2 = { '<STR_LIT>' : t1_meta . internal , <EOL> '<STR_LIT>' : name } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2 ) <EOL> expected_subreqs [ '<STR_LIT>' ] . append ( name ) <EOL> expected_subreqs [ '<STR_LIT:POST>' ] . append ( name ) <EOL> name = '<STR_LIT>' <EOL> t2 = self . ts_iter . next ( ) <EOL> tx_objs [ name ] = self . _create_ondisk_files ( tx_df_mgr , name , policy , t2 ) <EOL> t2_type = self . ts_iter . next ( ) <EOL> metadata_1 = { '<STR_LIT>' : t2_type . internal , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t2_type . internal } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_1 ) <EOL> t2_meta = self . ts_iter . next ( ) <EOL> metadata_2 = { '<STR_LIT>' : t2_meta . internal , <EOL> '<STR_LIT>' : name } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2 ) <EOL> rx_objs [ name ] = self . _create_ondisk_files ( rx_df_mgr , name , policy , t2 ) <EOL> rx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2 ) <EOL> expected_subreqs [ '<STR_LIT:POST>' ] . append ( name ) <EOL> name = '<STR_LIT>' <EOL> t3 = self . ts_iter . next ( ) <EOL> tx_objs [ name ] = self . _create_ondisk_files ( tx_df_mgr , name , policy , t3 ) <EOL> t3_type = self . ts_iter . next ( ) <EOL> metadata_1 = { '<STR_LIT>' : t3_type . internal , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t3_type . internal } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_1 ) <EOL> t3_meta = self . ts_iter . next ( ) <EOL> metadata_2 = { '<STR_LIT>' : t3_meta . internal , <EOL> '<STR_LIT>' : name } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2 ) <EOL> rx_objs [ name ] = self . _create_ondisk_files ( rx_df_mgr , name , policy , t3 ) <EOL> metadata_2b = { '<STR_LIT>' : t3_meta . internal , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t3_type . internal } <EOL> rx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2b ) <EOL> name = '<STR_LIT>' <EOL> t4 = self . ts_iter . next ( ) <EOL> tx_objs [ name ] = self . _create_ondisk_files ( tx_df_mgr , name , policy , t4 ) <EOL> t4_type = self . ts_iter . next ( ) <EOL> t4_meta = self . ts_iter . next ( ) <EOL> metadata_2b = { '<STR_LIT>' : t4_meta . internal , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t4_type . internal } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2b ) <EOL> rx_objs [ name ] = self . _create_ondisk_files ( rx_df_mgr , name , policy , t4 ) <EOL> metadata_1 = { '<STR_LIT>' : t4_type . internal , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t4_type . internal } <EOL> rx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_1 ) <EOL> metadata_2 = { '<STR_LIT>' : t4_meta . internal , <EOL> '<STR_LIT>' : name } <EOL> rx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata_2 ) <EOL> name = '<STR_LIT>' <EOL> t5 = self . ts_iter . next ( ) <EOL> tx_objs [ name ] = self . _create_ondisk_files ( tx_df_mgr , name , policy , t5 ) <EOL> t5_type = self . ts_iter . next ( ) <EOL> t5_meta = self . ts_iter . next ( ) <EOL> metadata = { '<STR_LIT>' : t5_meta . internal , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : t5_type . internal } <EOL> tx_objs [ name ] [ <NUM_LIT:0> ] . write_metadata ( metadata ) <EOL> rx_objs [ name ] = self . _create_ondisk_files ( rx_df_mgr , name , policy , t5 ) <EOL> expected_subreqs [ '<STR_LIT:POST>' ] . append ( name ) <EOL> suffixes = set ( ) <EOL> for diskfiles in tx_objs . values ( ) : <EOL> for df in diskfiles : <EOL> suffixes . add ( os . path . basename ( os . path . dirname ( df . _datadir ) ) ) <EOL> job = { '<STR_LIT>' : self . device , <EOL> '<STR_LIT>' : self . partition , <EOL> '<STR_LIT>' : policy } <EOL> node = dict ( self . rx_node ) <EOL> node . update ( { '<STR_LIT:index>' : rx_node_index } ) <EOL> sender = ssync_sender . Sender ( self . daemon , node , job , suffixes ) <EOL> sender . connect , trace = self . make_connect_wrapper ( sender ) <EOL> success , in_sync_objs = sender ( ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( in_sync_objs ) , trace [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( success ) <EOL> results = self . _analyze_trace ( trace ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( results [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( results [ '<STR_LIT>' ] ) ) <EOL> for subreq in results . get ( '<STR_LIT>' ) : <EOL> obj = subreq [ '<STR_LIT:path>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:3> ] <EOL> method = subreq [ '<STR_LIT>' ] <EOL> self . assertTrue ( obj in expected_subreqs [ method ] , <EOL> '<STR_LIT>' <EOL> % ( method , obj , expected_subreqs [ method ] ) ) <EOL> expected_subreqs [ method ] . remove ( obj ) <EOL> if method == '<STR_LIT>' : <EOL> expected_body = '<STR_LIT>' % subreq [ '<STR_LIT:path>' ] <EOL> self . assertEqual ( expected_body , subreq [ '<STR_LIT:body>' ] ) <EOL> for _method , expected in expected_subreqs . items ( ) : <EOL> self . assertFalse ( expected , <EOL> '<STR_LIT>' <EOL> % ( _method , expected ) ) <EOL> self . assertFalse ( results [ '<STR_LIT>' ] ) <EOL> self . _verify_ondisk_files ( tx_objs , policy ) <EOL> for oname , rx_obj in rx_objs . items ( ) : <EOL> df = rx_obj [ <NUM_LIT:0> ] . open ( ) <EOL> metadata = df . get_metadata ( ) <EOL> self . assertEqual ( metadata [ '<STR_LIT>' ] , oname ) <EOL> self . assertEqual ( metadata [ '<STR_LIT:Content-Type>' ] , '<STR_LIT>' ) <EOL> tx_hashes = tx_df_mgr . get_hashes ( <EOL> self . device , self . partition , suffixes , policy ) <EOL> rx_hashes = rx_df_mgr . get_hashes ( <EOL> self . device , self . partition , suffixes , policy ) <EOL> self . assertEqual ( tx_hashes , rx_hashes ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
