The " b"server is currently unavailable. Please try again at a " b"later time.
'") else: last_line = ( b":ERROR: 503 b'The " b"server is currently unavailable. Please try again at a " b"later time.
'") self.assertEqual( self.body_lines(resp.body), [last_line]) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_SSYNC_calls_replication_lock(self): with mock.patch.object( self.controller._diskfile_router[POLICIES.legacy], 'replication_lock') as mocked_replication_lock: req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) mocked_replication_lock.assert_called_once_with('sda1', POLICIES.legacy, '1') def test_Receiver_with_default_storage_policy(self): req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') rcvr = ssync_receiver.Receiver(self.controller, req) body_lines = [chunk.strip() for chunk in rcvr() if chunk.strip()] self.assertEqual( body_lines, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(rcvr.policy, POLICIES[0]) def test_Receiver_with_storage_policy_index_header(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') rcvr = ssync_receiver.Receiver(self.controller, req) body_lines = [chunk.strip() for chunk in rcvr() if chunk.strip()] self.assertEqual( body_lines, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(rcvr.policy, POLICIES[1]) self.assertIsNone(rcvr.frag_index) def test_Receiver_with_bad_storage_policy_index_header(self): valid_indices = sorted([int(policy) for policy in POLICIES]) bad_index = valid_indices[-1] + 1 req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '0', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': bad_index}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') self.controller.logger = mock.MagicMock() with self.assertRaises(HTTPException) as caught: ssync_receiver.Receiver(self.controller, req) self.assertEqual('503 Service Unavailable', caught.exception.status) self.assertEqual(b'No policy with index 2', caught.exception.body) @unit.patch_policies() def test_Receiver_with_only_frag_index_header(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '7', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') rcvr = ssync_receiver.Receiver(self.controller, req) body_lines = [chunk.strip() for chunk in rcvr() if chunk.strip()] self.assertEqual( body_lines, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(rcvr.policy, POLICIES[1]) self.assertEqual(rcvr.frag_index, 7) @unit.patch_policies() def test_Receiver_with_only_node_index_header(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_NODE_INDEX': '7', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') rcvr = ssync_receiver.Receiver(self.controller, req) body_lines = [chunk.strip() for chunk in rcvr() if chunk.strip()] self.assertEqual( body_lines, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(rcvr.policy, POLICIES[1]) # we used to require the reconstructor to send the frag_index twice as # two different headers because of evolutionary reasons, now we ignore # node_index self.assertEqual(rcvr.frag_index, None) @unit.patch_policies() def test_Receiver_with_matched_indexes(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_NODE_INDEX': '7', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '7', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') rcvr = ssync_receiver.Receiver(self.controller, req) body_lines = [chunk.strip() for chunk in rcvr() if chunk.strip()] self.assertEqual( body_lines, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(rcvr.policy, POLICIES[1]) self.assertEqual(rcvr.frag_index, 7) @unit.patch_policies() def test_Receiver_with_invalid_indexes(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_NODE_INDEX': 'None', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': 'None', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual(resp.status_int, 400) @unit.patch_policies() def test_Receiver_with_mismatched_indexes(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_NODE_INDEX': '6', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '7', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') rcvr = ssync_receiver.Receiver(self.controller, req) body_lines = [chunk.strip() for chunk in rcvr() if chunk.strip()] self.assertEqual( body_lines, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(rcvr.policy, POLICIES[1]) # node_index if provided should always match frag_index; but if they # differ, frag_index takes precedence self.assertEqual(rcvr.frag_index, 7) def test_SSYNC_replication_lock_fail(self): def _mock(path, policy, partition): with exceptions.ReplicationLockTimeout(0.01, '/somewhere/' + path): eventlet.sleep(0.05) with mock.patch.object( self.controller._diskfile_router[POLICIES.legacy], 'replication_lock', _mock): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b":ERROR: 0 '0.01 seconds: /somewhere/sda1'"]) self.controller.logger.debug.assert_called_once_with( 'None/sda1/1 SSYNC LOCK TIMEOUT: 0.01 seconds: ' '/somewhere/sda1') def test_SSYNC_replication_lock_per_partition(self): def _concurrent_ssync(path1, path2): env = {'REQUEST_METHOD': 'SSYNC'} body = ':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' \ ':UPDATES: START\r\n:UPDATES: END\r\n' req1 = swob.Request.blank(path1, environ=env, body=body) req2 = swob.Request.blank(path2, environ=env, body=body) rcvr1 = ssync_receiver.Receiver(self.controller, req1) rcvr2 = ssync_receiver.Receiver(self.controller, req2) body_lines1 = [] body_lines2 = [] for chunk1, chunk2 in six.moves.zip_longest(rcvr1(), rcvr2()): if chunk1 and chunk1.strip(): body_lines1.append(chunk1.strip()) if chunk2 and chunk2.strip(): body_lines2.append(chunk2.strip()) return body_lines1, body_lines2 self.controller._diskfile_router[POLICIES[0]]\ .replication_lock_timeout = 0.01 self.controller._diskfile_router[POLICIES[0]]\ .replication_concurrency_per_device = 2 # It should be possible to lock two different partitions body_lines1, body_lines2 = _concurrent_ssync('/sda1/1', '/sda1/2') self.assertEqual( body_lines1, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual( body_lines2, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) # It should not be possible to lock the same partition twice body_lines1, body_lines2 = _concurrent_ssync('/sda1/1', '/sda1/1') self.assertEqual( body_lines1, [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertRegex( b''.join(body_lines2), br"^:ERROR: 0 '0\.0[0-9]+ seconds: " br"/.+/sda1/objects/1/.lock-replication'$") def test_SSYNC_initial_path(self): with mock.patch.object( self.controller, 'replication_semaphore') as \ mocked_replication_semaphore: req = swob.Request.blank( '/device', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b"Invalid path: /device"]) self.assertEqual(resp.status_int, 400) self.assertFalse(mocked_replication_semaphore.acquire.called) self.assertFalse(mocked_replication_semaphore.release.called) with mock.patch.object( self.controller, 'replication_semaphore') as \ mocked_replication_semaphore: req = swob.Request.blank( '/device/', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b"Invalid path: /device/"]) self.assertEqual(resp.status_int, 400) self.assertFalse(mocked_replication_semaphore.acquire.called) self.assertFalse(mocked_replication_semaphore.release.called) with mock.patch.object( self.controller, 'replication_semaphore') as \ mocked_replication_semaphore: req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual(resp.body, b'\r\n') self.assertEqual(resp.status_int, 200) mocked_replication_semaphore.acquire.assert_called_once_with(0) mocked_replication_semaphore.release.assert_called_once_with() error_lines = self.logger.get_lines_for_level('error') self.assertEqual(['ssync client disconnected'], error_lines) with mock.patch.object( self.controller, 'replication_semaphore') as \ mocked_replication_semaphore: req = swob.Request.blank( '/device/partition/junk', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b"Invalid path: /device/partition/junk"]) self.assertEqual(resp.status_int, 400) self.assertFalse(mocked_replication_semaphore.acquire.called) self.assertFalse(mocked_replication_semaphore.release.called) def test_SSYNC_mount_check_isdir(self): with mock.patch.object(self.controller, 'replication_semaphore'), \ mock.patch.object( self.controller._diskfile_router[POLICIES.legacy], 'mount_check', False), \ mock_check_drive(isdir=True) as mocks: req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual(resp.body, b'\r\n') self.assertEqual(resp.status_int, 200) self.assertEqual([], mocks['ismount'].call_args_list) error_lines = self.logger.get_lines_for_level('error') self.assertEqual(['ssync client disconnected'], error_lines) def test_SSYNC_mount_check(self): with mock.patch.object(self.controller, 'replication_semaphore'), \ mock.patch.object( self.controller._diskfile_router[POLICIES.legacy], 'mount_check', True), \ mock_check_drive(ismount=False) as mocks: req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b"There " b"was not enough space to save the resource. Drive: " b"device
"]) self.assertEqual(resp.status_int, 507) self.assertEqual([mock.call(os.path.join( self.controller._diskfile_router[POLICIES.legacy].devices, 'device'))], mocks['ismount'].call_args_list) mocks['ismount'].reset_mock() mocks['ismount'].return_value = True req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}) resp = req.get_response(self.controller) self.assertEqual(resp.body, b'\r\n') self.assertEqual(resp.status_int, 200) self.assertEqual([mock.call(os.path.join( self.controller._diskfile_router[POLICIES.legacy].devices, 'device'))] * 2, mocks['ismount'].call_args_list) error_lines = self.logger.get_lines_for_level('error') self.assertEqual(['ssync client disconnected'], error_lines) def test_SSYNC_Exception(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def get_socket(self): return self.mock_socket with mock.patch.object( ssync_receiver.eventlet.greenio, 'shutdown_safe') as \ mock_shutdown_safe: self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\nBad content is here') req.remote_addr = '1.2.3.4' mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'Got no headers for Bad content is here'"]) self.assertEqual(resp.status_int, 200) mock_shutdown_safe.assert_called_once_with( mock_wsgi_input.mock_socket) mock_wsgi_input.mock_socket.close.assert_called_once_with() self.controller.logger.exception.assert_called_once_with( '1.2.3.4/device/partition EXCEPTION in ssync.Receiver') def test_SSYNC_Exception_Exception(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def get_socket(self): return self.mock_socket with mock.patch.object( ssync_receiver.eventlet.greenio, 'shutdown_safe') as \ mock_shutdown_safe: self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\nBad content is here') req.remote_addr = mock.MagicMock() req.remote_addr.__str__ = mock.Mock( side_effect=Exception("can't stringify this")) mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END']) self.assertEqual(resp.status_int, 200) mock_shutdown_safe.assert_called_once_with( mock_wsgi_input.mock_socket) mock_wsgi_input.mock_socket.close.assert_called_once_with() self.controller.logger.exception.assert_called_once_with( 'EXCEPTION in ssync.Receiver') def test_MISSING_CHECK_timeout(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def readline(self, sizehint=-1): line = io.BytesIO.readline(self) if line.startswith(b'hash'): eventlet.sleep(0.1) return line def get_socket(self): return self.mock_socket self.controller.client_timeout = 0.01 with mock.patch.object( ssync_receiver.eventlet.greenio, 'shutdown_safe') as \ mock_shutdown_safe: self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' 'hash ts\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') req.remote_addr = '2.3.4.5' mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b":ERROR: 408 '0.01 seconds: missing_check line'"]) self.assertEqual(resp.status_int, 200) self.assertTrue(mock_shutdown_safe.called) self.controller.logger.error.assert_called_once_with( '2.3.4.5/sda1/1 TIMEOUT in ssync.Receiver: ' '0.01 seconds: missing_check line') def test_MISSING_CHECK_other_exception(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def readline(self, sizehint=-1): line = io.BytesIO.readline(self) if line.startswith(b'hash'): raise Exception('test exception') return line def get_socket(self): return self.mock_socket self.controller.client_timeout = 0.01 with mock.patch.object( ssync_receiver.eventlet.greenio, 'shutdown_safe') as \ mock_shutdown_safe: self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' 'hash ts\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') req.remote_addr = '3.4.5.6' mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b":ERROR: 0 'test exception'"]) self.assertEqual(resp.status_int, 200) self.assertTrue(mock_shutdown_safe.called) self.controller.logger.exception.assert_called_once_with( '3.4.5.6/sda1/1 EXCEPTION in ssync.Receiver') def test_MISSING_CHECK_empty_list(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_have_none(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + self.ts1 + '\r\n' + self.hash2 + ' ' + self.ts2 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' dm').encode('ascii'), (self.hash2 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_extra_line_parts(self): # check that rx tolerates extra parts in missing check lines to # allow for protocol upgrades extra_1 = 'extra' extra_2 = 'multiple extra parts' self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + self.ts1 + ' ' + extra_1 + '\r\n' + self.hash2 + ' ' + self.ts2 + ' ' + extra_2 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' dm').encode('ascii'), (self.hash2 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_have_one_exact(self): object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, self.ts1 + '.data'), 'w+') fp.write('1') fp.flush() self.metadata1['Content-Length'] = '1' diskfile.write_metadata(fp, self.metadata1) self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + self.ts1 + '\r\n' + self.hash2 + ' ' + self.ts2 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash2 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_missing_meta_expired_data(self): # verify that even when rx disk file has expired x-delete-at, it will # still be opened and checked for missing meta self.controller.logger = mock.MagicMock() ts1 = next(make_timestamp_iter()) df = self.controller.get_diskfile( 'sda1', '1', self.account1, self.container1, self.object1, POLICIES[0]) write_diskfile(df, ts1, extra_metadata={'X-Delete-At': 0}) # make a request - expect newer metadata to be wanted req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + ts1.internal + ' m:30d40\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b'c2519f265f9633e74f9b2fe3b9bec27d m', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) @patch_policies(with_ec_default=True) def test_MISSING_CHECK_missing_durable(self): # check that local non-durable frag is made durable if remote sends # same ts for same frag, but only if remote is durable self.controller.logger = mock.MagicMock() self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) # make rx disk file but don't commit it, so durable state is missing ts1 = next(make_timestamp_iter()).internal object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, ts1 + '#2.data'), 'w+') fp.write('1') fp.flush() metadata1 = { 'name': self.name1, 'X-Timestamp': ts1, 'Content-Length': '1'} diskfile.write_metadata(fp, metadata1) self.assertEqual([ts1 + '#2.data'], os.listdir(object_dir)) # sanity # offer same non-durable frag - expect no data to be wanted req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + ts1 + ' durable:no\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) # the local frag is still not durable... self.assertEqual([ts1 + '#2.data'], os.listdir(object_dir)) # offer same frag but durable - expect no data to be wanted req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + ts1 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) # the local frag is now durable... self.assertEqual([ts1 + '#2#d.data'], os.listdir(object_dir)) @patch_policies(with_ec_default=True) @mock.patch('swift.obj.diskfile.ECDiskFileWriter.commit') def test_MISSING_CHECK_missing_durable_but_commit_fails(self, mock_commit): self.controller.logger = mock.MagicMock() self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) # make rx disk file but don't commit it, so durable state is missing ts1 = next(make_timestamp_iter()).internal object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, ts1 + '#2.data'), 'w+') fp.write('1') fp.flush() metadata1 = { 'name': self.name1, 'X-Timestamp': ts1, 'Content-Length': '1'} diskfile.write_metadata(fp, metadata1) self.assertEqual([ts1 + '#2.data'], os.listdir(object_dir)) # sanity # make a request with commit disabled - expect data to be wanted req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + ts1 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) # make a request with commit raising error - expect data to be wanted mock_commit.side_effect = Exception req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + ts1 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertTrue(self.controller.logger.exception.called) self.assertIn( 'EXCEPTION in ssync.Receiver while attempting commit of', self.controller.logger.exception.call_args[0][0]) @patch_policies(with_ec_default=True) def test_MISSING_CHECK_local_non_durable(self): # check that local non-durable fragment does not prevent other frags # being wanted from the sender self.controller.logger = mock.MagicMock() self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) ts_iter = make_timestamp_iter() ts1 = next(ts_iter).internal ts2 = next(ts_iter).internal ts3 = next(ts_iter).internal # make non-durable rx disk file at ts2 object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, ts2 + '#2.data'), 'w+') fp.write('1') fp.flush() metadata1 = { 'name': self.name1, 'X-Timestamp': ts2, 'Content-Length': '1'} diskfile.write_metadata(fp, metadata1) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) # sanity def do_check(tx_missing_line, expected_rx_missing_lines): req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + tx_missing_line + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START'] + [l.encode('ascii') for l in expected_rx_missing_lines] + [b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) # check remote frag is always wanted - older, newer, durable or not... do_check(self.hash1 + ' ' + ts1 + ' durable:no', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts1 + ' durable:yes', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts1, [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts3 + ' durable:no', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts3 + ' durable:yes', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts3, [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) # ... except when at same timestamp do_check(self.hash1 + ' ' + ts2 + ' durable:no', []) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) # durable remote frag at ts2 will make the local durable.. do_check(self.hash1 + ' ' + ts2 + ' durable:yes', []) self.assertEqual([ts2 + '#2#d.data'], os.listdir(object_dir)) @patch_policies(with_ec_default=True) def test_MISSING_CHECK_local_durable(self): # check that local durable fragment does not prevent newer non-durable # frags being wanted from the sender self.controller.logger = mock.MagicMock() self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) ts_iter = make_timestamp_iter() ts1 = next(ts_iter).internal ts2 = next(ts_iter).internal ts3 = next(ts_iter).internal # make non-durable rx disk file at ts2 object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, ts2 + '#2.data'), 'w+') fp.write('1') fp.flush() metadata1 = { 'name': self.name1, 'X-Timestamp': ts2, 'Content-Length': '1'} diskfile.write_metadata(fp, metadata1) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) # sanity def do_check(tx_missing_line, expected_rx_missing_lines): req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + tx_missing_line + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START'] + [l.encode('ascii') for l in expected_rx_missing_lines] + [b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) # check remote frag is always wanted - older, newer, durable or not... do_check(self.hash1 + ' ' + ts1 + ' durable:no', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts1 + ' durable:yes', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts1, [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts3 + ' durable:no', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts3 + ' durable:yes', [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) do_check(self.hash1 + ' ' + ts3, [self.hash1 + ' dm']) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) # ... except when at same timestamp do_check(self.hash1 + ' ' + ts2 + ' durable:no', []) self.assertEqual([ts2 + '#2.data'], os.listdir(object_dir)) # durable remote frag at ts2 will make the local durable.. do_check(self.hash1 + ' ' + ts2 + ' durable:yes', []) self.assertEqual([ts2 + '#2#d.data'], os.listdir(object_dir)) @patch_policies(with_ec_default=True) def test_MISSING_CHECK_local_durable_older_than_remote_non_durable(self): # check that newer non-durable fragment is wanted self.controller.logger = mock.MagicMock() self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) ts_iter = make_timestamp_iter() ts1 = next(ts_iter).internal ts2 = next(ts_iter).internal # make durable rx disk file at ts2 object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, ts1 + '#2#d.data'), 'w+') fp.write('1') fp.flush() metadata1 = { 'name': self.name1, 'X-Timestamp': ts1, 'Content-Length': '1'} diskfile.write_metadata(fp, metadata1) # make a request offering non-durable at ts2 req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '2'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + ts2 + ' durable:no\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_storage_policy(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[1])), '1', self.hash1) utils.mkdirs(object_dir) fp = open(os.path.join(object_dir, self.ts1 + '.data'), 'w+') fp.write('1') fp.flush() self.metadata1['Content-Length'] = '1' diskfile.write_metadata(fp, self.metadata1) self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + self.ts1 + '\r\n' + self.hash2 + ' ' + self.ts2 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash2 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_have_one_newer(self): object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) newer_ts1 = utils.normalize_timestamp(float(self.ts1) + 1) self.metadata1['X-Timestamp'] = newer_ts1 fp = open(os.path.join(object_dir, newer_ts1 + '.data'), 'w+') fp.write('1') fp.flush() self.metadata1['Content-Length'] = '1' diskfile.write_metadata(fp, self.metadata1) self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + self.ts1 + '\r\n' + self.hash2 + ' ' + self.ts2 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash2 + ' dm').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_have_newer_meta(self): object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) older_ts1 = utils.normalize_timestamp(float(self.ts1) - 1) self.metadata1['X-Timestamp'] = older_ts1 fp = open(os.path.join(object_dir, older_ts1 + '.data'), 'w+') fp.write('1') fp.flush() self.metadata1['Content-Length'] = '1' diskfile.write_metadata(fp, self.metadata1) # write newer .meta file metadata = {'name': self.name1, 'X-Timestamp': self.ts2, 'X-Object-Meta-Test': 'test'} fp = open(os.path.join(object_dir, self.ts2 + '.meta'), 'w+') diskfile.write_metadata(fp, metadata) # receiver has .data at older_ts, .meta at ts2 # sender has .data at ts1 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + self.ts1 + '\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' d').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_MISSING_CHECK_have_older_meta(self): object_dir = utils.storage_directory( os.path.join(self.testdir, 'sda1', diskfile.get_data_dir(POLICIES[0])), '1', self.hash1) utils.mkdirs(object_dir) older_ts1 = utils.normalize_timestamp(float(self.ts1) - 1) self.metadata1['X-Timestamp'] = older_ts1 fp = open(os.path.join(object_dir, older_ts1 + '.data'), 'w+') fp.write('1') fp.flush() self.metadata1['Content-Length'] = '1' diskfile.write_metadata(fp, self.metadata1) # write .meta file at ts1 metadata = {'name': self.name1, 'X-Timestamp': self.ts1, 'X-Object-Meta-Test': 'test'} fp = open(os.path.join(object_dir, self.ts1 + '.meta'), 'w+') diskfile.write_metadata(fp, metadata) # receiver has .data at older_ts, .meta at ts1 # sender has .data at older_ts, .meta at ts2 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/sda1/1', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n' + self.hash1 + ' ' + older_ts1 + ' m:30d40\r\n' ':MISSING_CHECK: END\r\n' ':UPDATES: START\r\n:UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', (self.hash1 + ' m').encode('ascii'), b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.error.called) self.assertFalse(self.controller.logger.exception.called) def test_UPDATES_no_start(self): # verify behavior when the sender disconnects and does not send # ':UPDATES: START' e.g. if a sender timeout pops while waiting for # receiver response to missing checks self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n') req.remote_addr = '2.3.4.5' resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END']) self.assertEqual(resp.status_int, 200) self.controller.logger.error.assert_called_once_with( 'ssync client disconnected') def test_UPDATES_timeout(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def readline(self, sizehint=-1): line = io.BytesIO.readline(self) if line.startswith(b'DELETE'): eventlet.sleep(0.1) return line def get_socket(self): return self.mock_socket self.controller.client_timeout = 0.01 with mock.patch.object( ssync_receiver.eventlet.greenio, 'shutdown_safe') as \ mock_shutdown_safe: self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'X-Timestamp: 1364456113.76334\r\n' '\r\n' ':UPDATES: END\r\n') req.remote_addr = '2.3.4.5' mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 408 '0.01 seconds: updates line'"]) self.assertEqual(resp.status_int, 200) mock_shutdown_safe.assert_called_once_with( mock_wsgi_input.mock_socket) mock_wsgi_input.mock_socket.close.assert_called_once_with() self.controller.logger.error.assert_called_once_with( '2.3.4.5/device/partition TIMEOUT in ssync.Receiver: ' '0.01 seconds: updates line') def test_UPDATES_other_exception(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def readline(self, sizehint=-1): line = io.BytesIO.readline(self) if line.startswith(b'DELETE'): raise Exception('test exception') return line def get_socket(self): return self.mock_socket self.controller.client_timeout = 0.01 with mock.patch.object( ssync_receiver.eventlet.greenio, 'shutdown_safe') as \ mock_shutdown_safe: self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'X-Timestamp: 1364456113.76334\r\n' '\r\n' ':UPDATES: END\r\n') req.remote_addr = '3.4.5.6' mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'test exception'"]) self.assertEqual(resp.status_int, 200) mock_shutdown_safe.assert_called_once_with( mock_wsgi_input.mock_socket) mock_wsgi_input.mock_socket.close.assert_called_once_with() self.controller.logger.exception.assert_called_once_with( '3.4.5.6/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_no_problems_no_hard_disconnect(self): class _Wrapper(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) self.mock_socket = mock.MagicMock() def get_socket(self): return self.mock_socket self.controller.client_timeout = 0.01 with mock.patch.object(ssync_receiver.eventlet.greenio, 'shutdown_safe') as mock_shutdown_safe, \ mock.patch.object( self.controller, 'DELETE', return_value=swob.HTTPNoContent()): req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'X-Timestamp: 1364456113.76334\r\n' '\r\n' ':UPDATES: END\r\n') mock_wsgi_input = _Wrapper(req.body) req.environ['wsgi.input'] = mock_wsgi_input resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(mock_shutdown_safe.called) self.assertFalse(mock_wsgi_input.mock_socket.close.called) def test_UPDATES_bad_subrequest_line(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'bad_subrequest_line\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', UNPACK_ERR]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') with mock.patch.object( self.controller, 'DELETE', return_value=swob.HTTPNoContent()): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'X-Timestamp: 1364456113.76334\r\n' '\r\n' 'bad_subrequest_line2') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', UNPACK_ERR]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_no_headers(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'Got no headers for DELETE /a/c/o'"]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_bad_headers(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'Bad-Header Test\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', UNPACK_ERR]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'Good-Header: Test\r\n' 'Bad-Header Test\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', UNPACK_ERR]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_bad_content_length(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o\r\n' 'Content-Length: a\r\n\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':ERROR: 0 "invalid literal for int() with base 10: \'a\'"']) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_content_length_with_DELETE(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'Content-Length: 1\r\n\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'DELETE subrequest with content-length /a/c/o'"]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_no_content_length_with_PUT(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o\r\n\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'No content-length sent for PUT /a/c/o'"]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') def test_UPDATES_early_termination(self): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o\r\n' 'Content-Length: 1\r\n\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END']) self.assertEqual(resp.status_int, 200) self.controller.logger.error.assert_called_once_with( 'None/device/partition read failed in ssync.Receiver: ' 'Early termination for PUT /a/c/o') def test_UPDATES_failures(self): @server.public def _DELETE(request): if request.path == '/device/partition/a/c/works': return swob.HTTPNoContent() else: return swob.HTTPInternalServerError() # failures never hit threshold with mock.patch.object(self.controller, 'DELETE', _DELETE): self.controller.replication_failure_threshold = 4 self.controller.replication_failure_ratio = 1.5 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n') resp = req.get_response(self.controller) if six.PY2: final_line = (b":ERROR: 500 'ERROR: With :UPDATES: " b"3 failures to 0 successes'") else: final_line = (b":ERROR: 500 b'ERROR: With :UPDATES: " b"3 failures to 0 successes'") self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', final_line]) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertTrue(self.controller.logger.warning.called) self.assertEqual(3, self.controller.logger.warning.call_count) self.controller.logger.clear() # failures hit threshold and no successes, so ratio is like infinity with mock.patch.object(self.controller, 'DELETE', _DELETE): self.controller.replication_failure_threshold = 4 self.controller.replication_failure_ratio = 1.5 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' ':UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'Too many 4 failures to 0 successes'"]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') self.assertFalse(self.controller.logger.error.called) self.assertTrue(self.controller.logger.warning.called) self.assertEqual(4, self.controller.logger.warning.call_count) self.controller.logger.clear() # failures hit threshold and ratio hits 1.33333333333 with mock.patch.object(self.controller, 'DELETE', _DELETE): self.controller.replication_failure_threshold = 4 self.controller.replication_failure_ratio = 1.5 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/works\r\n\r\n' 'DELETE /a/c/works\r\n\r\n' 'DELETE /a/c/works\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' ':UPDATES: END\r\n') resp = req.get_response(self.controller) if six.PY2: final_line = (b":ERROR: 500 'ERROR: With :UPDATES: " b"4 failures to 3 successes'") else: final_line = (b":ERROR: 500 b'ERROR: With :UPDATES: " b"4 failures to 3 successes'") self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', final_line]) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertTrue(self.controller.logger.warning.called) self.assertEqual(4, self.controller.logger.warning.call_count) self.controller.logger.clear() # failures hit threshold and ratio hits 2.0 with mock.patch.object(self.controller, 'DELETE', _DELETE): self.controller.replication_failure_threshold = 4 self.controller.replication_failure_ratio = 1.5 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/works\r\n\r\n' 'DELETE /a/c/works\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' 'DELETE /a/c/o\r\n\r\n' ':UPDATES: END\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'Too many 4 failures to 2 successes'"]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') self.assertFalse(self.controller.logger.error.called) self.assertTrue(self.controller.logger.warning.called) self.assertEqual(4, self.controller.logger.warning.call_count) self.controller.logger.clear() def test_UPDATES_PUT(self): _PUT_request = [None] @server.public def _PUT(request): _PUT_request[0] = request request.read_body = request.environ['wsgi.input'].read() return swob.HTTPCreated() with mock.patch.object(self.controller, 'PUT', _PUT): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=b':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' b':UPDATES: START\r\n' b'PUT /a/c/o\r\n' b'Content-Length: 1\r\n' b'Etag: c4ca4238a0b923820dcc509a6f75849b\r\n' b'X-Timestamp: 1364456113.12344\r\n' b'X-Object-Meta-Test1: one\r\n' b'X-Object-Meta-T\xc3\xa8st2: m\xc3\xa8ta\r\n' b'Content-Encoding: gzip\r\n' b'Specialty-Header: value\r\n' b'X-Backend-No-Commit: True\r\n' b'\r\n' b'1') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertEqual(len(_PUT_request), 1) # sanity req = _PUT_request[0] self.assertEqual(req.path, '/device/partition/a/c/o') self.assertEqual(req.content_length, 1) expected = { 'Etag': 'c4ca4238a0b923820dcc509a6f75849b', 'Content-Length': '1', 'X-Timestamp': '1364456113.12344', 'X-Object-Meta-Test1': 'one', 'X-Object-Meta-T\xc3\xa8st2': 'm\xc3\xa8ta', 'Content-Encoding': 'gzip', 'Specialty-Header': 'value', 'X-Backend-No-Commit': 'True', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', # note: Etag and X-Backend-No-Commit not in replication-headers 'X-Backend-Replication-Headers': ( 'content-length x-timestamp x-object-meta-test1 ' 'x-object-meta-t\xc3\xa8st2 content-encoding ' 'specialty-header')} self.assertEqual({k: req.headers[k] for k in expected}, expected) def test_UPDATES_PUT_replication_headers(self): self.controller.logger = mock.MagicMock() # sanity check - regular PUT will not persist Specialty-Header or # X-Backend-No-Commit req = swob.Request.blank( '/sda1/0/a/c/o1', body='1', environ={'REQUEST_METHOD': 'PUT'}, headers={'Content-Length': '1', 'Content-Type': 'text/plain', 'Etag': 'c4ca4238a0b923820dcc509a6f75849b', 'X-Timestamp': '1364456113.12344', 'X-Object-Meta-Test1': 'one', 'Content-Encoding': 'gzip', 'X-Backend-No-Commit': 'False', 'Specialty-Header': 'value'}) resp = req.get_response(self.controller) self.assertEqual(resp.status_int, 201) df = self.controller.get_diskfile( 'sda1', '0', 'a', 'c', 'o1', POLICIES.default) df.open() self.assertFalse('Specialty-Header' in df.get_metadata()) self.assertFalse('X-Backend-No-Commit' in df.get_metadata()) # an SSYNC request can override PUT header filtering... req = swob.Request.blank( '/sda1/0', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o2\r\n' 'Content-Length: 1\r\n' 'Content-Type: text/plain\r\n' 'Etag: c4ca4238a0b923820dcc509a6f75849b\r\n' 'X-Timestamp: 1364456113.12344\r\n' 'X-Object-Meta-Test1: one\r\n' 'Content-Encoding: gzip\r\n' 'X-Backend-No-Commit: False\r\n' 'Specialty-Header: value\r\n' '\r\n' '1') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) # verify diskfile has metadata permitted by replication headers # including Specialty-Header, but not Etag or X-Backend-No-Commit df = self.controller.get_diskfile( 'sda1', '0', 'a', 'c', 'o2', POLICIES.default) df.open() for chunk in df.reader(): self.assertEqual(b'1', chunk) expected = {'ETag': 'c4ca4238a0b923820dcc509a6f75849b', 'Content-Length': '1', 'Content-Type': 'text/plain', 'X-Timestamp': '1364456113.12344', 'X-Object-Meta-Test1': 'one', 'Content-Encoding': 'gzip', 'Specialty-Header': 'value', 'name': '/a/c/o2'} actual = df.get_metadata() self.assertEqual(expected, actual) def test_UPDATES_POST(self): _POST_request = [None] @server.public def _POST(request): _POST_request[0] = request return swob.HTTPAccepted() with mock.patch.object(self.controller, 'POST', _POST): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'POST /a/c/o\r\n' 'X-Timestamp: 1364456113.12344\r\n' 'X-Object-Meta-Test1: one\r\n' 'Specialty-Header: value\r\n\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) req = _POST_request[0] self.assertEqual(req.path, '/device/partition/a/c/o') self.assertIsNone(req.content_length) self.assertEqual(req.headers, { 'X-Timestamp': '1364456113.12344', 'X-Object-Meta-Test1': 'one', 'Specialty-Header': 'value', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'x-timestamp x-object-meta-test1 specialty-header')}) def test_UPDATES_with_storage_policy(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) _PUT_request = [None] @server.public def _PUT(request): _PUT_request[0] = request request.read_body = request.environ['wsgi.input'].read() return swob.HTTPCreated() with mock.patch.object(self.controller, 'PUT', _PUT): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '1'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o\r\n' 'Content-Length: 1\r\n' 'X-Timestamp: 1364456113.12344\r\n' 'X-Object-Meta-Test1: one\r\n' 'Content-Encoding: gzip\r\n' 'Specialty-Header: value\r\n' '\r\n' '1') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertEqual(len(_PUT_request), 1) # sanity req = _PUT_request[0] self.assertEqual(req.path, '/device/partition/a/c/o') self.assertEqual(req.content_length, 1) self.assertEqual(req.headers, { 'Content-Length': '1', 'X-Timestamp': '1364456113.12344', 'X-Object-Meta-Test1': 'one', 'Content-Encoding': 'gzip', 'Specialty-Header': 'value', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '1', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp x-object-meta-test1 ' 'content-encoding specialty-header')}) self.assertEqual(req.read_body, b'1') def test_UPDATES_PUT_with_storage_policy_and_node_index(self): # update router post policy patch self.controller._diskfile_router = diskfile.DiskFileRouter( self.conf, self.controller.logger) _PUT_request = [None] @server.public def _PUT(request): _PUT_request[0] = request request.read_body = request.environ['wsgi.input'].read() return swob.HTTPCreated() with mock.patch.object(self.controller, 'PUT', _PUT): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC', 'HTTP_X_BACKEND_SSYNC_NODE_INDEX': '7', 'HTTP_X_BACKEND_SSYNC_FRAG_INDEX': '7', 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX': '0'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o\r\n' 'Content-Length: 1\r\n' 'X-Timestamp: 1364456113.12344\r\n' 'X-Object-Meta-Test1: one\r\n' 'Content-Encoding: gzip\r\n' 'Specialty-Header: value\r\n' '\r\n' '1') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertEqual(len(_PUT_request), 1) # sanity req = _PUT_request[0] self.assertEqual(req.path, '/device/partition/a/c/o') self.assertEqual(req.content_length, 1) self.assertEqual(req.headers, { 'Content-Length': '1', 'X-Timestamp': '1364456113.12344', 'X-Object-Meta-Test1': 'one', 'Content-Encoding': 'gzip', 'Specialty-Header': 'value', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Ssync-Frag-Index': '7', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp x-object-meta-test1 ' 'content-encoding specialty-header')}) self.assertEqual(req.read_body, b'1') def test_UPDATES_DELETE(self): _DELETE_request = [None] @server.public def _DELETE(request): _DELETE_request[0] = request return swob.HTTPNoContent() with mock.patch.object(self.controller, 'DELETE', _DELETE): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'DELETE /a/c/o\r\n' 'X-Timestamp: 1364456113.76334\r\n' '\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertEqual(len(_DELETE_request), 1) # sanity req = _DELETE_request[0] self.assertEqual(req.path, '/device/partition/a/c/o') self.assertEqual(req.headers, { 'X-Timestamp': '1364456113.76334', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': 'x-timestamp'}) def test_UPDATES_BONK(self): _BONK_request = [None] @server.public def _BONK(request): _BONK_request[0] = request return swob.HTTPOk() self.controller.BONK = _BONK self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'BONK /a/c/o\r\n' 'X-Timestamp: 1364456113.76334\r\n' '\r\n') resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b":ERROR: 0 'Invalid subrequest method BONK'"]) self.assertEqual(resp.status_int, 200) self.controller.logger.exception.assert_called_once_with( 'None/device/partition EXCEPTION in ssync.Receiver') self.assertEqual(len(_BONK_request), 1) # sanity self.assertIsNone(_BONK_request[0]) def test_UPDATES_multiple(self): _requests = [] @server.public def _PUT(request): _requests.append(request) request.read_body = request.environ['wsgi.input'].read() return swob.HTTPCreated() @server.public def _POST(request): _requests.append(request) return swob.HTTPOk() @server.public def _DELETE(request): _requests.append(request) return swob.HTTPNoContent() with mock.patch.object(self.controller, 'PUT', _PUT), \ mock.patch.object(self.controller, 'POST', _POST), \ mock.patch.object(self.controller, 'DELETE', _DELETE): self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o1\r\n' 'Content-Length: 1\r\n' 'X-Timestamp: 1364456113.00001\r\n' 'X-Object-Meta-Test1: one\r\n' 'Content-Encoding: gzip\r\n' 'Specialty-Header: value\r\n' '\r\n' '1' 'DELETE /a/c/o2\r\n' 'X-Timestamp: 1364456113.00002\r\n' '\r\n' 'PUT /a/c/o3\r\n' 'Content-Length: 3\r\n' 'X-Timestamp: 1364456113.00003\r\n' '\r\n' '123' 'PUT /a/c/o4\r\n' 'Content-Length: 4\r\n' 'X-Timestamp: 1364456113.00004\r\n' '\r\n' '1\r\n4' 'DELETE /a/c/o5\r\n' 'X-Timestamp: 1364456113.00005\r\n' '\r\n' 'DELETE /a/c/o6\r\n' 'X-Timestamp: 1364456113.00006\r\n' '\r\n' 'PUT /a/c/o7\r\n' 'Content-Length: 7\r\n' 'X-Timestamp: 1364456113.00007\r\n' '\r\n' '1234567' 'POST /a/c/o7\r\n' 'X-Object-Meta-Test-User: user_meta\r\n' 'X-Timestamp: 1364456113.00008\r\n' '\r\n' ) resp = req.get_response(self.controller) self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', b':UPDATES: START', b':UPDATES: END']) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertEqual(len(_requests), 8) # sanity req = _requests.pop(0) self.assertEqual(req.method, 'PUT') self.assertEqual(req.path, '/device/partition/a/c/o1') self.assertEqual(req.content_length, 1) self.assertEqual(req.headers, { 'Content-Length': '1', 'X-Timestamp': '1364456113.00001', 'X-Object-Meta-Test1': 'one', 'Content-Encoding': 'gzip', 'Specialty-Header': 'value', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp x-object-meta-test1 ' 'content-encoding specialty-header')}) self.assertEqual(req.read_body, b'1') req = _requests.pop(0) self.assertEqual(req.method, 'DELETE') self.assertEqual(req.path, '/device/partition/a/c/o2') self.assertEqual(req.headers, { 'X-Timestamp': '1364456113.00002', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': 'x-timestamp'}) req = _requests.pop(0) self.assertEqual(req.method, 'PUT') self.assertEqual(req.path, '/device/partition/a/c/o3') self.assertEqual(req.content_length, 3) self.assertEqual(req.headers, { 'Content-Length': '3', 'X-Timestamp': '1364456113.00003', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp')}) self.assertEqual(req.read_body, b'123') req = _requests.pop(0) self.assertEqual(req.method, 'PUT') self.assertEqual(req.path, '/device/partition/a/c/o4') self.assertEqual(req.content_length, 4) self.assertEqual(req.headers, { 'Content-Length': '4', 'X-Timestamp': '1364456113.00004', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp')}) self.assertEqual(req.read_body, b'1\r\n4') req = _requests.pop(0) self.assertEqual(req.method, 'DELETE') self.assertEqual(req.path, '/device/partition/a/c/o5') self.assertEqual(req.headers, { 'X-Timestamp': '1364456113.00005', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': 'x-timestamp'}) req = _requests.pop(0) self.assertEqual(req.method, 'DELETE') self.assertEqual(req.path, '/device/partition/a/c/o6') self.assertEqual(req.headers, { 'X-Timestamp': '1364456113.00006', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': 'x-timestamp'}) req = _requests.pop(0) self.assertEqual(req.method, 'PUT') self.assertEqual(req.path, '/device/partition/a/c/o7') self.assertEqual(req.content_length, 7) self.assertEqual(req.headers, { 'Content-Length': '7', 'X-Timestamp': '1364456113.00007', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp')}) self.assertEqual(req.read_body, b'1234567') req = _requests.pop(0) self.assertEqual(req.method, 'POST') self.assertEqual(req.path, '/device/partition/a/c/o7') self.assertIsNone(req.content_length) self.assertEqual(req.headers, { 'X-Timestamp': '1364456113.00008', 'X-Object-Meta-Test-User': 'user_meta', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'x-object-meta-test-user x-timestamp')}) self.assertEqual(_requests, []) def test_UPDATES_subreq_does_not_read_all(self): # This tests that if a SSYNC subrequest fails and doesn't read # all the subrequest body that it will read and throw away the rest of # the body before moving on to the next subrequest. # If you comment out the part in ssync_receiver where it does: # for junk in subreq.environ['wsgi.input']: # pass # You can then see this test fail. _requests = [] @server.public def _PUT(request): _requests.append(request) # Deliberately just reading up to first 2 bytes. request.read_body = request.environ['wsgi.input'].read(2) return swob.HTTPInternalServerError() class _IgnoreReadlineHint(io.BytesIO): def __init__(self, value): io.BytesIO.__init__(self, value) def readline(self, hint=-1): return io.BytesIO.readline(self) self.controller.PUT = _PUT self.controller.network_chunk_size = 2 self.controller.logger = mock.MagicMock() req = swob.Request.blank( '/device/partition', environ={'REQUEST_METHOD': 'SSYNC'}, body=':MISSING_CHECK: START\r\n:MISSING_CHECK: END\r\n' ':UPDATES: START\r\n' 'PUT /a/c/o1\r\n' 'Content-Length: 3\r\n' 'X-Timestamp: 1364456113.00001\r\n' '\r\n' '123' 'PUT /a/c/o2\r\n' 'Content-Length: 1\r\n' 'X-Timestamp: 1364456113.00002\r\n' '\r\n' '1') req.environ['wsgi.input'] = _IgnoreReadlineHint(req.body) resp = req.get_response(self.controller) if six.PY2: final_line = (b":ERROR: 500 'ERROR: With :UPDATES: " b"2 failures to 0 successes'") else: final_line = (b":ERROR: 500 b'ERROR: With :UPDATES: " b"2 failures to 0 successes'") self.assertEqual( self.body_lines(resp.body), [b':MISSING_CHECK: START', b':MISSING_CHECK: END', final_line]) self.assertEqual(resp.status_int, 200) self.assertFalse(self.controller.logger.exception.called) self.assertFalse(self.controller.logger.error.called) self.assertTrue(self.controller.logger.warning.called) self.assertEqual(2, self.controller.logger.warning.call_count) self.assertEqual(len(_requests), 2) # sanity req = _requests.pop(0) self.assertEqual(req.path, '/device/partition/a/c/o1') self.assertEqual(req.content_length, 3) self.assertEqual(req.headers, { 'Content-Length': '3', 'X-Timestamp': '1364456113.00001', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp')}) self.assertEqual(req.read_body, b'12') req = _requests.pop(0) self.assertEqual(req.path, '/device/partition/a/c/o2') self.assertEqual(req.content_length, 1) self.assertEqual(req.headers, { 'Content-Length': '1', 'X-Timestamp': '1364456113.00002', 'Host': 'localhost:80', 'X-Backend-Storage-Policy-Index': '0', 'X-Backend-Replication': 'True', 'X-Backend-Replication-Headers': ( 'content-length x-timestamp')}) self.assertEqual(req.read_body, b'1') self.assertEqual(_requests, []) @patch_policies(with_ec_default=True) class TestSsyncRxServer(unittest.TestCase): # Tests to verify behavior of SSYNC requests sent to an object # server socket. def setUp(self): skip_if_no_xattrs() # dirs self.tmpdir = tempfile.mkdtemp() self.tempdir = os.path.join(self.tmpdir, 'tmp_test_obj_server') self.devices = os.path.join(self.tempdir, 'srv/node') for device in ('sda1', 'sdb1'): os.makedirs(os.path.join(self.devices, device)) self.conf = { 'devices': self.devices, 'mount_check': 'false', 'swift_dir': self.tempdir, } self.rx_logger = debug_logger('test-object-server') rx_server = server.ObjectController(self.conf, logger=self.rx_logger) self.rx_ip = '127.0.0.1' self.sock = listen_zero() self.rx_server = eventlet.spawn( eventlet.wsgi.server, self.sock, rx_server, utils.NullLogger()) self.rx_port = self.sock.getsockname()[1] self.tx_logger = debug_logger('test-reconstructor') self.daemon = ObjectReconstructor(self.conf, self.tx_logger) self.daemon._diskfile_mgr = self.daemon._df_router[POLICIES[0]] def tearDown(self): self.rx_server.kill() self.sock.close() eventlet.sleep(0) shutil.rmtree(self.tmpdir) def test_SSYNC_disconnect(self): node = { 'replication_ip': '127.0.0.1', 'replication_port': self.rx_port, 'device': 'sdb1', } job = { 'partition': 0, 'policy': POLICIES[0], 'device': 'sdb1', } sender = ssync_sender.Sender(self.daemon, node, job, ['abc']) # kick off the sender and let the error trigger failure with mock.patch( 'swift.obj.ssync_receiver.Receiver.initialize_request') \ as mock_initialize_request: mock_initialize_request.side_effect = \ swob.HTTPInternalServerError() success, _ = sender() self.assertFalse(success) stderr = six.StringIO() with mock.patch('sys.stderr', stderr): # let gc and eventlet spin a bit del sender for i in range(3): eventlet.sleep(0) self.assertNotIn('ValueError: invalid literal for int() with base 16', stderr.getvalue()) def test_SSYNC_device_not_available(self): with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check')\ as mock_missing_check: self.connection = bufferedhttp.BufferedHTTPConnection( '127.0.0.1:%s' % self.rx_port) self.connection.putrequest('SSYNC', '/sdc1/0') self.connection.putheader('Transfer-Encoding', 'chunked') self.connection.putheader('X-Backend-Storage-Policy-Index', int(POLICIES[0])) self.connection.endheaders() resp = self.connection.getresponse() self.assertEqual(507, resp.status) resp.read() resp.close() # sanity check that the receiver did not proceed to missing_check self.assertFalse(mock_missing_check.called) def test_SSYNC_read_error(self): # verify that read errors from wsgi reader are caught and reported def do_send(data): self.rx_logger.clear() self.connection = bufferedhttp.BufferedHTTPConnection( '127.0.0.1:%s' % self.rx_port) self.connection.putrequest('SSYNC', '/sda1/0') self.connection.putheader('Transfer-Encoding', 'chunked') self.connection.putheader('X-Backend-Storage-Policy-Index', int(POLICIES[0])) self.connection.endheaders() resp = self.connection.getresponse() self.assertEqual(200, resp.status) resp.close() self.connection.send(data) self.connection.close() for sleep_time in (0, 0.1, 1): lines = self.rx_logger.get_lines_for_level('error') if lines: return lines eventlet.sleep(sleep_time) return [] # check read errors during missing_check phase error_lines = do_send(b'') self.assertEqual(1, len(error_lines)) self.assertIn('missing_check start: invalid literal', error_lines[0]) error_lines = do_send(b'1\r\n') self.assertEqual(1, len(error_lines)) self.assertIn('missing_check start: unexpected end of file', error_lines[0]) error_lines = do_send(b'17\r\n:MISSING_CHECK: START\r\n\r\nx\r\n') self.assertEqual(1, len(error_lines)) self.assertIn('missing_check line: invalid literal', error_lines[0]) error_lines = do_send(b'17\r\n:MISSING_CHECK: START\r\n\r\n12\r\n') self.assertEqual(1, len(error_lines)) self.assertIn('missing_check line: unexpected end of file', error_lines[0]) # check read errors during updates phase with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check'): error_lines = do_send(b'') self.assertEqual(1, len(error_lines)) self.assertIn('updates start: invalid literal', error_lines[0]) with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check'): error_lines = do_send(b'1\r\n') self.assertEqual(1, len(error_lines)) self.assertIn('updates start: unexpected end of file', error_lines[0]) with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check'): error_lines = do_send(b'11\r\n:UPDATES: START\r\n\r\nx\r\n') self.assertEqual(1, len(error_lines)) self.assertIn('updates line: invalid literal', error_lines[0]) with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check'): error_lines = do_send(b'11\r\n:UPDATES: START\r\n\r\n12\r\n') self.assertEqual(1, len(error_lines)) self.assertIn('updates line: unexpected end of file', error_lines[0]) def test_SSYNC_invalid_policy(self): valid_indices = sorted([int(policy) for policy in POLICIES]) bad_index = valid_indices[-1] + 1 with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check')\ as mock_missing_check: self.connection = bufferedhttp.BufferedHTTPConnection( '127.0.0.1:%s' % self.rx_port) self.connection.putrequest('SSYNC', '/sda1/0') self.connection.putheader('Transfer-Encoding', 'chunked') self.connection.putheader('X-Backend-Storage-Policy-Index', bad_index) self.connection.endheaders() resp = self.connection.getresponse() self.assertEqual(503, resp.status) resp.read() resp.close() # sanity check that the receiver did not proceed to missing_check self.assertFalse(mock_missing_check.called) def test_bad_request_invalid_frag_index(self): with mock.patch('swift.obj.ssync_receiver.Receiver.missing_check')\ as mock_missing_check: self.connection = bufferedhttp.BufferedHTTPConnection( '127.0.0.1:%s' % self.rx_port) self.connection.putrequest('SSYNC', '/sda1/0') self.connection.putheader('Transfer-Encoding', 'chunked') self.connection.putheader('X-Backend-Ssync-Frag-Index', 'None') self.connection.endheaders() resp = self.connection.getresponse() self.assertEqual(400, resp.status) error_msg = resp.read() self.assertIn(b"Invalid X-Backend-Ssync-Frag-Index 'None'", error_msg) resp.close() # sanity check that the receiver did not proceed to missing_check self.assertFalse(mock_missing_check.called) class TestModuleMethods(unittest.TestCase): def test_decode_missing(self): object_hash = '9d41d8cd98f00b204e9800998ecf0abc' ts_iter = make_timestamp_iter() t_data = next(ts_iter) t_meta = next(ts_iter) t_ctype = next(ts_iter) d_meta_data = t_meta.raw - t_data.raw d_ctype_data = t_ctype.raw - t_data.raw # legacy single timestamp string msg = '%s %s' % (object_hash, t_data.internal) expected = dict(object_hash=object_hash, ts_meta=t_data, ts_data=t_data, ts_ctype=t_data, durable=True) self.assertEqual(expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # hex meta delta encoded as extra message part msg = '%s %s m:%x' % (object_hash, t_data.internal, d_meta_data) expected = dict(object_hash=object_hash, ts_data=t_data, ts_meta=t_meta, ts_ctype=t_data, durable=True) self.assertEqual(expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # hex content type delta encoded in extra message part msg = '%s %s t:%x,m:%x' % (object_hash, t_data.internal, d_ctype_data, d_meta_data) expected = dict(object_hash=object_hash, ts_data=t_data, ts_meta=t_meta, ts_ctype=t_ctype, durable=True) self.assertEqual( expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # order of subparts does not matter msg = '%s %s m:%x,t:%x' % (object_hash, t_data.internal, d_meta_data, d_ctype_data) self.assertEqual( expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # timestamps have offsets t_data_offset = utils.Timestamp(t_data, offset=99) t_meta_offset = utils.Timestamp(t_meta, offset=1) t_ctype_offset = utils.Timestamp(t_ctype, offset=2) expected = dict(object_hash=object_hash, ts_data=t_data_offset, ts_meta=t_meta_offset, ts_ctype=t_ctype_offset, durable=True) expected = ('%s %s_0000000000000063 m:%x__1,t:%x__2' % (object_hash, t_data.internal, d_meta_data, d_ctype_data)) self.assertEqual( expected.encode('ascii'), ssync_sender.encode_missing( object_hash, t_data_offset, t_meta_offset, t_ctype_offset, durable=True)) # hex content type delta may be zero msg = '%s %s t:0,m:%x' % (object_hash, t_data.internal, d_meta_data) expected = dict(object_hash=object_hash, ts_data=t_data, ts_meta=t_meta, ts_ctype=t_data, durable=True) self.assertEqual( expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # unexpected zero delta is tolerated msg = '%s %s m:0' % (object_hash, t_data.internal) expected = dict(object_hash=object_hash, ts_meta=t_data, ts_data=t_data, ts_ctype=t_data, durable=True) self.assertEqual(expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # unexpected subparts in timestamp delta part are tolerated msg = '%s %s c:12345,m:%x,junk' % (object_hash, t_data.internal, d_meta_data) expected = dict(object_hash=object_hash, ts_meta=t_meta, ts_data=t_data, ts_ctype=t_data, durable=True) self.assertEqual( expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # extra message parts tolerated msg = '%s %s m:%x future parts' % (object_hash, t_data.internal, d_meta_data) expected = dict(object_hash=object_hash, ts_meta=t_meta, ts_data=t_data, ts_ctype=t_data, durable=True) self.assertEqual(expected, ssync_receiver.decode_missing(msg.encode('ascii'))) # not durable def check_non_durable(durable_val): msg = '%s %s m:%x,durable:%s' % (object_hash, t_data.internal, d_meta_data, durable_val) expected = dict(object_hash=object_hash, ts_meta=t_meta, ts_data=t_data, ts_ctype=t_data, durable=False) self.assertEqual( expected, ssync_receiver.decode_missing(msg.encode('ascii'))) check_non_durable('no') check_non_durable('false') check_non_durable('False') # explicit durable (as opposed to True by default) def check_durable(durable_val): msg = '%s %s m:%x,durable:%s' % (object_hash, t_data.internal, d_meta_data, durable_val) expected = dict(object_hash=object_hash, ts_meta=t_meta, ts_data=t_data, ts_ctype=t_data, durable=True) self.assertEqual( expected, ssync_receiver.decode_missing(msg.encode('ascii'))) check_durable('yes') check_durable('true') check_durable('True') def test_encode_wanted(self): ts_iter = make_timestamp_iter() old_t_data = next(ts_iter) t_data = next(ts_iter) old_t_meta = next(ts_iter) t_meta = next(ts_iter) remote = { 'object_hash': 'theremotehash', 'ts_data': t_data, 'ts_meta': t_meta, } # missing local = {} expected = 'theremotehash dm' self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) # in-sync local = { 'ts_data': t_data, 'ts_meta': t_meta, } expected = None self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) # out-of-sync local = { 'ts_data': old_t_data, 'ts_meta': old_t_meta, } expected = 'theremotehash dm' self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) # old data local = { 'ts_data': old_t_data, 'ts_meta': t_meta, } expected = 'theremotehash d' self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) # old metadata local = { 'ts_data': t_data, 'ts_meta': old_t_meta, } expected = 'theremotehash m' self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) # in-sync tombstone local = { 'ts_data': t_data, } expected = None self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) # old tombstone local = { 'ts_data': old_t_data, } expected = 'theremotehash d' self.assertEqual(ssync_receiver.encode_wanted(remote, local), expected) if __name__ == '__main__': unittest.main()