7624b198cf
FakeLogger gets better log level handling Parameterize logger on some daemons which were previously unparameterized and try and use the interface in tests. FakeRing use more real code The existing FakeRing mock's implementation bit me on some pretty subtle character encoding issue by-passing the hash_path code that is normally part of get_part_nodes. This change tries to exercise more of the real ring code paths when it makes sense and provide a better Fake for use in testing. Add write_fake_ring helper to test.unit for when you need a real ring. DocImpact Implements: blueprint storage-policies Change-Id: Id2e3740b1dd569050f4e083617e7dd6a4249027e
1228 lines
50 KiB
Python
1228 lines
50 KiB
Python
# Copyright (c) 2010-2012 OpenStack Foundation
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
import os
|
|
import logging
|
|
import errno
|
|
import math
|
|
import time
|
|
from mock import patch, call
|
|
from shutil import rmtree
|
|
from tempfile import mkdtemp, NamedTemporaryFile
|
|
import mock
|
|
import simplejson
|
|
|
|
from swift.container.backend import DATADIR
|
|
from swift.common import db_replicator
|
|
from swift.common.utils import normalize_timestamp
|
|
from swift.common.exceptions import DriveNotMounted
|
|
from swift.common.swob import HTTPException
|
|
|
|
from test import unit
|
|
|
|
|
|
TEST_ACCOUNT_NAME = 'a c t'
|
|
TEST_CONTAINER_NAME = 'c o n'
|
|
|
|
|
|
def teardown_module():
|
|
"clean up my monkey patching"
|
|
reload(db_replicator)
|
|
|
|
|
|
@contextmanager
|
|
def lock_parent_directory(filename):
|
|
yield True
|
|
|
|
|
|
class FakeRing(object):
|
|
class Ring(object):
|
|
devs = []
|
|
|
|
def __init__(self, path, reload_time=15, ring_name=None):
|
|
pass
|
|
|
|
def get_part(self, account, container=None, obj=None):
|
|
return 0
|
|
|
|
def get_part_nodes(self, part):
|
|
return []
|
|
|
|
def get_more_nodes(self, *args):
|
|
return []
|
|
|
|
|
|
class FakeRingWithSingleNode(object):
|
|
class Ring(object):
|
|
devs = [dict(
|
|
id=1, weight=10.0, zone=1, ip='1.1.1.1', port=6000, device='sdb',
|
|
meta='', replication_ip='1.1.1.1', replication_port=6000
|
|
)]
|
|
|
|
def __init__(self, path, reload_time=15, ring_name=None):
|
|
pass
|
|
|
|
def get_part(self, account, container=None, obj=None):
|
|
return 0
|
|
|
|
def get_part_nodes(self, part):
|
|
return self.devs
|
|
|
|
def get_more_nodes(self, *args):
|
|
return (d for d in self.devs)
|
|
|
|
|
|
class FakeRingWithNodes(object):
|
|
class Ring(object):
|
|
devs = [dict(
|
|
id=1, weight=10.0, zone=1, ip='1.1.1.1', port=6000, device='sdb',
|
|
meta=''
|
|
), dict(
|
|
id=2, weight=10.0, zone=2, ip='1.1.1.2', port=6000, device='sdb',
|
|
meta=''
|
|
), dict(
|
|
id=3, weight=10.0, zone=3, ip='1.1.1.3', port=6000, device='sdb',
|
|
meta=''
|
|
), dict(
|
|
id=4, weight=10.0, zone=4, ip='1.1.1.4', port=6000, device='sdb',
|
|
meta=''
|
|
), dict(
|
|
id=5, weight=10.0, zone=5, ip='1.1.1.5', port=6000, device='sdb',
|
|
meta=''
|
|
), dict(
|
|
id=6, weight=10.0, zone=6, ip='1.1.1.6', port=6000, device='sdb',
|
|
meta='')]
|
|
|
|
def __init__(self, path, reload_time=15, ring_name=None):
|
|
pass
|
|
|
|
def get_part(self, account, container=None, obj=None):
|
|
return 0
|
|
|
|
def get_part_nodes(self, part):
|
|
return self.devs[:3]
|
|
|
|
def get_more_nodes(self, *args):
|
|
return (d for d in self.devs[3:])
|
|
|
|
|
|
class FakeProcess(object):
|
|
def __init__(self, *codes):
|
|
self.codes = iter(codes)
|
|
self.args = None
|
|
self.kwargs = None
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
class Failure(object):
|
|
def communicate(innerself):
|
|
next = self.codes.next()
|
|
if isinstance(next, int):
|
|
innerself.returncode = next
|
|
return next
|
|
raise next
|
|
return Failure()
|
|
|
|
|
|
@contextmanager
|
|
def _mock_process(*args):
|
|
orig_process = db_replicator.subprocess.Popen
|
|
db_replicator.subprocess.Popen = FakeProcess(*args)
|
|
yield db_replicator.subprocess.Popen
|
|
db_replicator.subprocess.Popen = orig_process
|
|
|
|
|
|
class ReplHttp(object):
|
|
def __init__(self, response=None, set_status=200):
|
|
self.response = response
|
|
self.set_status = set_status
|
|
replicated = False
|
|
host = 'localhost'
|
|
|
|
def replicate(self, *args):
|
|
self.replicated = True
|
|
|
|
class Response(object):
|
|
status = self.set_status
|
|
data = self.response
|
|
|
|
def read(innerself):
|
|
return self.response
|
|
return Response()
|
|
|
|
|
|
class ChangingMtimesOs(object):
|
|
def __init__(self):
|
|
self.mtime = 0
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
self.mtime += 1
|
|
return self.mtime
|
|
|
|
|
|
class FakeBroker(object):
|
|
db_file = __file__
|
|
get_repl_missing_table = False
|
|
stub_replication_info = None
|
|
db_type = 'container'
|
|
info = {'account': TEST_ACCOUNT_NAME, 'container': TEST_CONTAINER_NAME}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.locked = False
|
|
return None
|
|
|
|
@contextmanager
|
|
def lock(self):
|
|
self.locked = True
|
|
yield True
|
|
self.locked = False
|
|
|
|
def get_sync(self, *args, **kwargs):
|
|
return 5
|
|
|
|
def get_syncs(self):
|
|
return []
|
|
|
|
def get_items_since(self, point, *args):
|
|
if point == 0:
|
|
return [{'ROWID': 1}]
|
|
if point == -1:
|
|
return [{'ROWID': 1}, {'ROWID': 2}]
|
|
return []
|
|
|
|
def merge_syncs(self, *args, **kwargs):
|
|
self.args = args
|
|
|
|
def merge_items(self, *args):
|
|
self.args = args
|
|
|
|
def get_replication_info(self):
|
|
if self.get_repl_missing_table:
|
|
raise Exception('no such table')
|
|
if self.stub_replication_info:
|
|
return self.stub_replication_info
|
|
return {'delete_timestamp': 0, 'put_timestamp': 1, 'count': 0,
|
|
'hash': 12345, 'created_at': 1}
|
|
|
|
def reclaim(self, item_timestamp, sync_timestamp):
|
|
pass
|
|
|
|
def get_info(self):
|
|
return self.info
|
|
|
|
def newid(self, remote_d):
|
|
pass
|
|
|
|
def update_metadata(self, metadata):
|
|
self.metadata = metadata
|
|
|
|
def merge_timestamps(self, created_at, put_timestamp, delete_timestamp):
|
|
self.created_at = created_at
|
|
self.put_timestamp = put_timestamp
|
|
self.delete_timestamp = delete_timestamp
|
|
|
|
|
|
class FakeAccountBroker(FakeBroker):
|
|
db_type = 'account'
|
|
info = {'account': TEST_ACCOUNT_NAME}
|
|
|
|
|
|
class TestReplicator(db_replicator.Replicator):
|
|
server_type = 'container'
|
|
ring_file = 'container.ring.gz'
|
|
brokerclass = FakeBroker
|
|
datadir = DATADIR
|
|
default_port = 1000
|
|
|
|
|
|
class TestDBReplicator(unittest.TestCase):
|
|
def setUp(self):
|
|
db_replicator.ring = FakeRing()
|
|
self.delete_db_calls = []
|
|
self._patchers = []
|
|
|
|
def tearDown(self):
|
|
for patcher in self._patchers:
|
|
patcher.stop()
|
|
|
|
def _patch(self, patching_fn, *args, **kwargs):
|
|
patcher = patching_fn(*args, **kwargs)
|
|
patched_thing = patcher.start()
|
|
self._patchers.append(patcher)
|
|
return patched_thing
|
|
|
|
def stub_delete_db(self, object_file):
|
|
self.delete_db_calls.append(object_file)
|
|
|
|
def test_repl_connection(self):
|
|
node = {'replication_ip': '127.0.0.1', 'replication_port': 80,
|
|
'device': 'sdb1'}
|
|
conn = db_replicator.ReplConnection(node, '1234567890', 'abcdefg',
|
|
logging.getLogger())
|
|
|
|
def req(method, path, body, headers):
|
|
self.assertEquals(method, 'REPLICATE')
|
|
self.assertEquals(headers['Content-Type'], 'application/json')
|
|
|
|
class Resp(object):
|
|
def read(self):
|
|
return 'data'
|
|
resp = Resp()
|
|
conn.request = req
|
|
conn.getresponse = lambda *args: resp
|
|
self.assertEquals(conn.replicate(1, 2, 3), resp)
|
|
|
|
def other_req(method, path, body, headers):
|
|
raise Exception('blah')
|
|
conn.request = other_req
|
|
self.assertEquals(conn.replicate(1, 2, 3), None)
|
|
|
|
def test_rsync_file(self):
|
|
replicator = TestReplicator({})
|
|
with _mock_process(-1):
|
|
self.assertEquals(
|
|
False,
|
|
replicator._rsync_file('/some/file', 'remote:/some/file'))
|
|
with _mock_process(0):
|
|
self.assertEquals(
|
|
True,
|
|
replicator._rsync_file('/some/file', 'remote:/some/file'))
|
|
|
|
def test_rsync_file_popen_args(self):
|
|
replicator = TestReplicator({})
|
|
with _mock_process(0) as process:
|
|
replicator._rsync_file('/some/file', 'remote:/some_file')
|
|
exp_args = ([
|
|
'rsync', '--quiet', '--no-motd',
|
|
'--timeout=%s' % int(math.ceil(replicator.node_timeout)),
|
|
'--contimeout=%s' % int(math.ceil(replicator.conn_timeout)),
|
|
'--whole-file', '/some/file', 'remote:/some_file'],)
|
|
self.assertEqual(exp_args, process.args)
|
|
|
|
def test_rsync_file_popen_args_whole_file_false(self):
|
|
replicator = TestReplicator({})
|
|
with _mock_process(0) as process:
|
|
replicator._rsync_file('/some/file', 'remote:/some_file', False)
|
|
exp_args = ([
|
|
'rsync', '--quiet', '--no-motd',
|
|
'--timeout=%s' % int(math.ceil(replicator.node_timeout)),
|
|
'--contimeout=%s' % int(math.ceil(replicator.conn_timeout)),
|
|
'/some/file', 'remote:/some_file'],)
|
|
self.assertEqual(exp_args, process.args)
|
|
|
|
def test_rsync_db(self):
|
|
replicator = TestReplicator({})
|
|
replicator._rsync_file = lambda *args: True
|
|
fake_device = {'replication_ip': '127.0.0.1', 'device': 'sda1'}
|
|
replicator._rsync_db(FakeBroker(), fake_device, ReplHttp(), 'abcd')
|
|
|
|
def test_rsync_db_rsync_file_call(self):
|
|
fake_device = {'ip': '127.0.0.1', 'port': '0',
|
|
'replication_ip': '127.0.0.1', 'replication_port': '0',
|
|
'device': 'sda1'}
|
|
|
|
def mock_rsync_ip(ip):
|
|
self.assertEquals(fake_device['ip'], ip)
|
|
return 'rsync_ip(%s)' % ip
|
|
|
|
class MyTestReplicator(TestReplicator):
|
|
def __init__(self, db_file, remote_file):
|
|
super(MyTestReplicator, self).__init__({})
|
|
self.db_file = db_file
|
|
self.remote_file = remote_file
|
|
|
|
def _rsync_file(self_, db_file, remote_file, whole_file=True):
|
|
self.assertEqual(self_.db_file, db_file)
|
|
self.assertEqual(self_.remote_file, remote_file)
|
|
self_._rsync_file_called = True
|
|
return False
|
|
|
|
with patch('swift.common.db_replicator.rsync_ip', mock_rsync_ip):
|
|
broker = FakeBroker()
|
|
remote_file = 'rsync_ip(127.0.0.1)::container/sda1/tmp/abcd'
|
|
replicator = MyTestReplicator(broker.db_file, remote_file)
|
|
replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
|
|
self.assert_(replicator._rsync_file_called)
|
|
|
|
with patch('swift.common.db_replicator.rsync_ip', mock_rsync_ip):
|
|
broker = FakeBroker()
|
|
remote_file = 'rsync_ip(127.0.0.1)::container0/sda1/tmp/abcd'
|
|
replicator = MyTestReplicator(broker.db_file, remote_file)
|
|
replicator.vm_test_mode = True
|
|
replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
|
|
self.assert_(replicator._rsync_file_called)
|
|
|
|
def test_rsync_db_rsync_file_failure(self):
|
|
class MyTestReplicator(TestReplicator):
|
|
def __init__(self):
|
|
super(MyTestReplicator, self).__init__({})
|
|
self._rsync_file_called = False
|
|
|
|
def _rsync_file(self_, *args, **kwargs):
|
|
self.assertEqual(
|
|
False, self_._rsync_file_called,
|
|
'_sync_file() should only be called once')
|
|
self_._rsync_file_called = True
|
|
return False
|
|
|
|
with patch('os.path.exists', lambda *args: True):
|
|
replicator = MyTestReplicator()
|
|
fake_device = {'ip': '127.0.0.1', 'replication_ip': '127.0.0.1',
|
|
'device': 'sda1'}
|
|
replicator._rsync_db(FakeBroker(), fake_device, ReplHttp(), 'abcd')
|
|
self.assertEqual(True, replicator._rsync_file_called)
|
|
|
|
def test_rsync_db_change_after_sync(self):
|
|
class MyTestReplicator(TestReplicator):
|
|
def __init__(self, broker):
|
|
super(MyTestReplicator, self).__init__({})
|
|
self.broker = broker
|
|
self._rsync_file_call_count = 0
|
|
|
|
def _rsync_file(self_, db_file, remote_file, whole_file=True):
|
|
self_._rsync_file_call_count += 1
|
|
if self_._rsync_file_call_count == 1:
|
|
self.assertEquals(True, whole_file)
|
|
self.assertEquals(False, self_.broker.locked)
|
|
elif self_._rsync_file_call_count == 2:
|
|
self.assertEquals(False, whole_file)
|
|
self.assertEquals(True, self_.broker.locked)
|
|
else:
|
|
raise RuntimeError('_rsync_file() called too many times')
|
|
return True
|
|
|
|
# with journal file
|
|
with patch('os.path.exists', lambda *args: True):
|
|
broker = FakeBroker()
|
|
replicator = MyTestReplicator(broker)
|
|
fake_device = {'ip': '127.0.0.1', 'replication_ip': '127.0.0.1',
|
|
'device': 'sda1'}
|
|
replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
|
|
self.assertEquals(2, replicator._rsync_file_call_count)
|
|
|
|
# with new mtime
|
|
with patch('os.path.exists', lambda *args: False):
|
|
with patch('os.path.getmtime', ChangingMtimesOs()):
|
|
broker = FakeBroker()
|
|
replicator = MyTestReplicator(broker)
|
|
fake_device = {'ip': '127.0.0.1',
|
|
'replication_ip': '127.0.0.1',
|
|
'device': 'sda1'}
|
|
replicator._rsync_db(broker, fake_device, ReplHttp(), 'abcd')
|
|
self.assertEquals(2, replicator._rsync_file_call_count)
|
|
|
|
def test_in_sync(self):
|
|
replicator = TestReplicator({})
|
|
self.assertEquals(replicator._in_sync(
|
|
{'id': 'a', 'point': 0, 'max_row': 0, 'hash': 'b'},
|
|
{'id': 'a', 'point': -1, 'max_row': 0, 'hash': 'b'},
|
|
FakeBroker(), -1), True)
|
|
self.assertEquals(replicator._in_sync(
|
|
{'id': 'a', 'point': -1, 'max_row': 0, 'hash': 'b'},
|
|
{'id': 'a', 'point': -1, 'max_row': 10, 'hash': 'b'},
|
|
FakeBroker(), -1), True)
|
|
self.assertEquals(bool(replicator._in_sync(
|
|
{'id': 'a', 'point': -1, 'max_row': 0, 'hash': 'c'},
|
|
{'id': 'a', 'point': -1, 'max_row': 10, 'hash': 'd'},
|
|
FakeBroker(), -1)), False)
|
|
|
|
def test_run_once(self):
|
|
replicator = TestReplicator({})
|
|
replicator.run_once()
|
|
|
|
def test_run_once_no_ips(self):
|
|
replicator = TestReplicator({}, logger=unit.FakeLogger())
|
|
self._patch(patch.object, db_replicator, 'whataremyips',
|
|
lambda *args: [])
|
|
|
|
replicator.run_once()
|
|
|
|
self.assertEqual(
|
|
replicator.logger.log_dict['error'],
|
|
[(('ERROR Failed to get my own IPs?',), {})])
|
|
|
|
def test_run_once_node_is_not_mounted(self):
|
|
db_replicator.ring = FakeRingWithSingleNode()
|
|
conf = {'mount_check': 'true', 'bind_port': 6000}
|
|
replicator = TestReplicator(conf, logger=unit.FakeLogger())
|
|
self.assertEqual(replicator.mount_check, True)
|
|
self.assertEqual(replicator.port, 6000)
|
|
|
|
def mock_ismount(path):
|
|
self.assertEquals(path,
|
|
os.path.join(replicator.root,
|
|
replicator.ring.devs[0]['device']))
|
|
return False
|
|
|
|
self._patch(patch.object, db_replicator, 'whataremyips',
|
|
lambda *args: ['1.1.1.1'])
|
|
self._patch(patch.object, db_replicator, 'ismount', mock_ismount)
|
|
replicator.run_once()
|
|
|
|
self.assertEqual(
|
|
replicator.logger.log_dict['warning'],
|
|
[(('Skipping %(device)s as it is not mounted' %
|
|
replicator.ring.devs[0],), {})])
|
|
|
|
def test_run_once_node_is_mounted(self):
|
|
db_replicator.ring = FakeRingWithSingleNode()
|
|
conf = {'mount_check': 'true', 'bind_port': 6000}
|
|
replicator = TestReplicator(conf, logger=unit.FakeLogger())
|
|
self.assertEqual(replicator.mount_check, True)
|
|
self.assertEqual(replicator.port, 6000)
|
|
|
|
def mock_unlink_older_than(path, mtime):
|
|
self.assertEquals(path,
|
|
os.path.join(replicator.root,
|
|
replicator.ring.devs[0]['device'],
|
|
'tmp'))
|
|
self.assertTrue(time.time() - replicator.reclaim_age >= mtime)
|
|
|
|
def mock_spawn_n(fn, part, object_file, node_id):
|
|
self.assertEquals('123', part)
|
|
self.assertEquals('/srv/node/sda/c.db', object_file)
|
|
self.assertEquals(1, node_id)
|
|
|
|
self._patch(patch.object, db_replicator, 'whataremyips',
|
|
lambda *args: ['1.1.1.1'])
|
|
self._patch(patch.object, db_replicator, 'ismount', lambda *args: True)
|
|
self._patch(patch.object, db_replicator, 'unlink_older_than',
|
|
mock_unlink_older_than)
|
|
self._patch(patch.object, db_replicator, 'roundrobin_datadirs',
|
|
lambda *args: [('123', '/srv/node/sda/c.db', 1)])
|
|
self._patch(patch.object, replicator.cpool, 'spawn_n', mock_spawn_n)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.isdir.return_value = True
|
|
replicator.run_once()
|
|
mock_os.path.isdir.assert_called_with(
|
|
os.path.join(replicator.root,
|
|
replicator.ring.devs[0]['device'],
|
|
replicator.datadir))
|
|
|
|
def test_usync(self):
|
|
fake_http = ReplHttp()
|
|
replicator = TestReplicator({})
|
|
replicator._usync_db(0, FakeBroker(), fake_http, '12345', '67890')
|
|
|
|
def test_usync_http_error_above_300(self):
|
|
fake_http = ReplHttp(set_status=301)
|
|
replicator = TestReplicator({})
|
|
self.assertFalse(
|
|
replicator._usync_db(0, FakeBroker(), fake_http, '12345', '67890'))
|
|
|
|
def test_usync_http_error_below_200(self):
|
|
fake_http = ReplHttp(set_status=101)
|
|
replicator = TestReplicator({})
|
|
self.assertFalse(
|
|
replicator._usync_db(0, FakeBroker(), fake_http, '12345', '67890'))
|
|
|
|
def test_stats(self):
|
|
# I'm not sure how to test that this logs the right thing,
|
|
# but we can at least make sure it gets covered.
|
|
replicator = TestReplicator({})
|
|
replicator._zero_stats()
|
|
replicator._report_stats()
|
|
|
|
def test_replicate_object(self):
|
|
db_replicator.ring = FakeRingWithNodes()
|
|
replicator = TestReplicator({})
|
|
replicator.delete_db = self.stub_delete_db
|
|
replicator._replicate_object('0', '/path/to/file', 'node_id')
|
|
self.assertEquals([], self.delete_db_calls)
|
|
|
|
def test_replicate_object_quarantine(self):
|
|
replicator = TestReplicator({})
|
|
self._patch(patch.object, replicator.brokerclass, 'db_file',
|
|
'/a/b/c/d/e/hey')
|
|
self._patch(patch.object, replicator.brokerclass,
|
|
'get_repl_missing_table', True)
|
|
|
|
def mock_renamer(was, new, cause_colision=False):
|
|
if cause_colision and '-' not in new:
|
|
raise OSError(errno.EEXIST, "File already exists")
|
|
self.assertEquals('/a/b/c/d/e', was)
|
|
if '-' in new:
|
|
self.assert_(
|
|
new.startswith('/a/quarantined/containers/e-'))
|
|
else:
|
|
self.assertEquals('/a/quarantined/containers/e', new)
|
|
|
|
def mock_renamer_error(was, new):
|
|
return mock_renamer(was, new, cause_colision=True)
|
|
with patch.object(db_replicator, 'renamer', mock_renamer):
|
|
replicator._replicate_object('0', 'file', 'node_id')
|
|
# try the double quarantine
|
|
with patch.object(db_replicator, 'renamer', mock_renamer_error):
|
|
replicator._replicate_object('0', 'file', 'node_id')
|
|
|
|
def test_replicate_object_delete_because_deleted(self):
|
|
replicator = TestReplicator({})
|
|
try:
|
|
replicator.delete_db = self.stub_delete_db
|
|
replicator.brokerclass.stub_replication_info = {
|
|
'delete_timestamp': 2, 'put_timestamp': 1, 'count': 0}
|
|
replicator._replicate_object('0', '/path/to/file', 'node_id')
|
|
finally:
|
|
replicator.brokerclass.stub_replication_info = None
|
|
self.assertEquals(['/path/to/file'], self.delete_db_calls)
|
|
|
|
def test_replicate_object_delete_because_not_shouldbehere(self):
|
|
replicator = TestReplicator({})
|
|
replicator.delete_db = self.stub_delete_db
|
|
replicator._replicate_object('0', '/path/to/file', 'node_id')
|
|
self.assertEquals(['/path/to/file'], self.delete_db_calls)
|
|
|
|
def test_replicate_account_out_of_place(self):
|
|
replicator = TestReplicator({}, logger=unit.FakeLogger())
|
|
replicator.ring = FakeRingWithNodes().Ring('path')
|
|
replicator.brokerclass = FakeAccountBroker
|
|
replicator._repl_to_node = lambda *args: True
|
|
replicator.delete_db = self.stub_delete_db
|
|
# Correct node_id, wrong part
|
|
part = replicator.ring.get_part(TEST_ACCOUNT_NAME) + 1
|
|
node_id = replicator.ring.get_part_nodes(part)[0]['id']
|
|
replicator._replicate_object(str(part), '/path/to/file', node_id)
|
|
self.assertEqual(['/path/to/file'], self.delete_db_calls)
|
|
self.assertEqual(
|
|
replicator.logger.log_dict['error'],
|
|
[(('Found /path/to/file for /a%20c%20t when it should be on '
|
|
'partition 0; will replicate out and remove.',), {})])
|
|
|
|
def test_replicate_container_out_of_place(self):
|
|
replicator = TestReplicator({}, logger=unit.FakeLogger())
|
|
replicator.ring = FakeRingWithNodes().Ring('path')
|
|
replicator._repl_to_node = lambda *args: True
|
|
replicator.delete_db = self.stub_delete_db
|
|
# Correct node_id, wrong part
|
|
part = replicator.ring.get_part(
|
|
TEST_ACCOUNT_NAME, TEST_CONTAINER_NAME) + 1
|
|
node_id = replicator.ring.get_part_nodes(part)[0]['id']
|
|
replicator._replicate_object(str(part), '/path/to/file', node_id)
|
|
self.assertEqual(['/path/to/file'], self.delete_db_calls)
|
|
self.assertEqual(
|
|
replicator.logger.log_dict['error'],
|
|
[(('Found /path/to/file for /a%20c%20t/c%20o%20n when it should '
|
|
'be on partition 0; will replicate out and remove.',), {})])
|
|
|
|
def test_delete_db(self):
|
|
db_replicator.lock_parent_directory = lock_parent_directory
|
|
replicator = TestReplicator({}, logger=unit.FakeLogger())
|
|
replicator._zero_stats()
|
|
replicator.extract_device = lambda _: 'some_device'
|
|
|
|
temp_dir = mkdtemp()
|
|
try:
|
|
temp_suf_dir = os.path.join(temp_dir, '16e')
|
|
os.mkdir(temp_suf_dir)
|
|
temp_hash_dir = os.path.join(temp_suf_dir,
|
|
'166e33924a08ede4204871468c11e16e')
|
|
os.mkdir(temp_hash_dir)
|
|
temp_file = NamedTemporaryFile(dir=temp_hash_dir, delete=False)
|
|
temp_hash_dir2 = os.path.join(temp_suf_dir,
|
|
'266e33924a08ede4204871468c11e16e')
|
|
os.mkdir(temp_hash_dir2)
|
|
temp_file2 = NamedTemporaryFile(dir=temp_hash_dir2, delete=False)
|
|
|
|
# sanity-checks
|
|
self.assertTrue(os.path.exists(temp_dir))
|
|
self.assertTrue(os.path.exists(temp_suf_dir))
|
|
self.assertTrue(os.path.exists(temp_hash_dir))
|
|
self.assertTrue(os.path.exists(temp_file.name))
|
|
self.assertTrue(os.path.exists(temp_hash_dir2))
|
|
self.assertTrue(os.path.exists(temp_file2.name))
|
|
self.assertEqual(0, replicator.stats['remove'])
|
|
|
|
replicator.delete_db(temp_file.name)
|
|
|
|
self.assertTrue(os.path.exists(temp_dir))
|
|
self.assertTrue(os.path.exists(temp_suf_dir))
|
|
self.assertFalse(os.path.exists(temp_hash_dir))
|
|
self.assertFalse(os.path.exists(temp_file.name))
|
|
self.assertTrue(os.path.exists(temp_hash_dir2))
|
|
self.assertTrue(os.path.exists(temp_file2.name))
|
|
self.assertEqual([(('removes.some_device',), {})],
|
|
replicator.logger.log_dict['increment'])
|
|
self.assertEqual(1, replicator.stats['remove'])
|
|
|
|
replicator.delete_db(temp_file2.name)
|
|
|
|
self.assertTrue(os.path.exists(temp_dir))
|
|
self.assertFalse(os.path.exists(temp_suf_dir))
|
|
self.assertFalse(os.path.exists(temp_hash_dir))
|
|
self.assertFalse(os.path.exists(temp_file.name))
|
|
self.assertFalse(os.path.exists(temp_hash_dir2))
|
|
self.assertFalse(os.path.exists(temp_file2.name))
|
|
self.assertEqual([(('removes.some_device',), {})] * 2,
|
|
replicator.logger.log_dict['increment'])
|
|
self.assertEqual(2, replicator.stats['remove'])
|
|
finally:
|
|
rmtree(temp_dir)
|
|
|
|
def test_extract_device(self):
|
|
replicator = TestReplicator({'devices': '/some/root'})
|
|
self.assertEqual('some_device', replicator.extract_device(
|
|
'/some/root/some_device/deeper/and/deeper'))
|
|
self.assertEqual('UNKNOWN', replicator.extract_device(
|
|
'/some/foo/some_device/deeper/and/deeper'))
|
|
|
|
# def test_dispatch(self):
|
|
# rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
# no_op = lambda *args, **kwargs: True
|
|
# self.assertEquals(rpc.dispatch(('drv', 'part', 'hash'), ('op',)
|
|
# ).status_int, 400)
|
|
# rpc.mount_check = True
|
|
# self.assertEquals(rpc.dispatch(('drv', 'part', 'hash'), ['op',]
|
|
# ).status_int, 507)
|
|
# rpc.mount_check = False
|
|
# rpc.rsync_then_merge = lambda drive, db_file,
|
|
# args: self.assertEquals(args, ['test1'])
|
|
# rpc.complete_rsync = lambda drive, db_file,
|
|
# args: self.assertEquals(args, ['test2'])
|
|
# rpc.dispatch(('drv', 'part', 'hash'), ['rsync_then_merge','test1'])
|
|
# rpc.dispatch(('drv', 'part', 'hash'), ['complete_rsync','test2'])
|
|
# rpc.dispatch(('drv', 'part', 'hash'), ['other_op',])
|
|
|
|
def test_dispatch_no_arg_pop(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
response = rpc.dispatch(('a',), 'arg')
|
|
self.assertEquals('Invalid object type', response.body)
|
|
self.assertEquals(400, response.status_int)
|
|
|
|
def test_dispatch_drive_not_mounted(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, True)
|
|
|
|
def mock_ismount(path):
|
|
self.assertEquals('/drive', path)
|
|
return False
|
|
|
|
self._patch(patch.object, db_replicator, 'ismount', mock_ismount)
|
|
|
|
response = rpc.dispatch(('drive', 'part', 'hash'), ['method'])
|
|
|
|
self.assertEquals('507 drive is not mounted', response.status)
|
|
self.assertEquals(507, response.status_int)
|
|
|
|
def test_dispatch_unexpected_operation_db_does_not_exist(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
def mock_mkdirs(path):
|
|
self.assertEquals('/drive/tmp', path)
|
|
|
|
self._patch(patch.object, db_replicator, 'mkdirs', mock_mkdirs)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = False
|
|
response = rpc.dispatch(('drive', 'part', 'hash'), ['unexpected'])
|
|
|
|
self.assertEquals('404 Not Found', response.status)
|
|
self.assertEquals(404, response.status_int)
|
|
|
|
def test_dispatch_operation_unexpected(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
self._patch(patch.object, db_replicator, 'mkdirs', lambda *args: True)
|
|
|
|
def unexpected_method(broker, args):
|
|
self.assertEquals(FakeBroker, broker.__class__)
|
|
self.assertEqual(['arg1', 'arg2'], args)
|
|
return 'unexpected-called'
|
|
|
|
rpc.unexpected = unexpected_method
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = True
|
|
response = rpc.dispatch(('drive', 'part', 'hash'),
|
|
['unexpected', 'arg1', 'arg2'])
|
|
mock_os.path.exists.assert_called_with('/part/ash/hash/hash.db')
|
|
|
|
self.assertEquals('unexpected-called', response)
|
|
|
|
def test_dispatch_operation_rsync_then_merge(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
self._patch(patch.object, db_replicator, 'renamer', lambda *args: True)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = True
|
|
response = rpc.dispatch(('drive', 'part', 'hash'),
|
|
['rsync_then_merge', 'arg1', 'arg2'])
|
|
expected_calls = [call('/part/ash/hash/hash.db'),
|
|
call('/drive/tmp/arg1')]
|
|
self.assertEquals(mock_os.path.exists.call_args_list,
|
|
expected_calls)
|
|
self.assertEquals('204 No Content', response.status)
|
|
self.assertEquals(204, response.status_int)
|
|
|
|
def test_dispatch_operation_complete_rsync(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
self._patch(patch.object, db_replicator, 'renamer', lambda *args: True)
|
|
|
|
with patch('swift.common.db_replicator.os', new=mock.MagicMock(
|
|
wraps=os)) as mock_os:
|
|
mock_os.path.exists.side_effect = [False, True]
|
|
response = rpc.dispatch(('drive', 'part', 'hash'),
|
|
['complete_rsync', 'arg1', 'arg2'])
|
|
expected_calls = [call('/part/ash/hash/hash.db'),
|
|
call('/drive/tmp/arg1')]
|
|
self.assertEquals(mock_os.path.exists.call_args_list,
|
|
expected_calls)
|
|
self.assertEquals('204 No Content', response.status)
|
|
self.assertEquals(204, response.status_int)
|
|
|
|
def test_rsync_then_merge_db_does_not_exist(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = False
|
|
response = rpc.rsync_then_merge('drive', '/data/db.db',
|
|
('arg1', 'arg2'))
|
|
mock_os.path.exists.assert_called_with('/data/db.db')
|
|
self.assertEquals('404 Not Found', response.status)
|
|
self.assertEquals(404, response.status_int)
|
|
|
|
def test_rsync_then_merge_old_does_not_exist(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.side_effect = [True, False]
|
|
response = rpc.rsync_then_merge('drive', '/data/db.db',
|
|
('arg1', 'arg2'))
|
|
expected_calls = [call('/data/db.db'), call('/drive/tmp/arg1')]
|
|
self.assertEquals(mock_os.path.exists.call_args_list,
|
|
expected_calls)
|
|
self.assertEquals('404 Not Found', response.status)
|
|
self.assertEquals(404, response.status_int)
|
|
|
|
def test_rsync_then_merge_with_objects(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
def mock_renamer(old, new):
|
|
self.assertEquals('/drive/tmp/arg1', old)
|
|
self.assertEquals('/data/db.db', new)
|
|
|
|
self._patch(patch.object, db_replicator, 'renamer', mock_renamer)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = True
|
|
response = rpc.rsync_then_merge('drive', '/data/db.db',
|
|
['arg1', 'arg2'])
|
|
self.assertEquals('204 No Content', response.status)
|
|
self.assertEquals(204, response.status_int)
|
|
|
|
def test_complete_rsync_db_does_not_exist(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = True
|
|
response = rpc.complete_rsync('drive', '/data/db.db',
|
|
['arg1', 'arg2'])
|
|
mock_os.path.exists.assert_called_with('/data/db.db')
|
|
self.assertEquals('404 Not Found', response.status)
|
|
self.assertEquals(404, response.status_int)
|
|
|
|
def test_complete_rsync_old_file_does_not_exist(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.return_value = False
|
|
response = rpc.complete_rsync('drive', '/data/db.db',
|
|
['arg1', 'arg2'])
|
|
expected_calls = [call('/data/db.db'), call('/drive/tmp/arg1')]
|
|
self.assertEquals(expected_calls,
|
|
mock_os.path.exists.call_args_list)
|
|
self.assertEquals('404 Not Found', response.status)
|
|
self.assertEquals(404, response.status_int)
|
|
|
|
def test_complete_rsync_rename(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
|
|
def mock_exists(path):
|
|
if path == '/data/db.db':
|
|
return False
|
|
self.assertEquals('/drive/tmp/arg1', path)
|
|
return True
|
|
|
|
def mock_renamer(old, new):
|
|
self.assertEquals('/drive/tmp/arg1', old)
|
|
self.assertEquals('/data/db.db', new)
|
|
|
|
self._patch(patch.object, db_replicator, 'renamer', mock_renamer)
|
|
|
|
with patch('swift.common.db_replicator.os',
|
|
new=mock.MagicMock(wraps=os)) as mock_os:
|
|
mock_os.path.exists.side_effect = [False, True]
|
|
response = rpc.complete_rsync('drive', '/data/db.db',
|
|
['arg1', 'arg2'])
|
|
self.assertEquals('204 No Content', response.status)
|
|
self.assertEquals(204, response.status_int)
|
|
|
|
def test_replicator_sync_with_broker_replication_missing_table(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
broker = FakeBroker()
|
|
broker.get_repl_missing_table = True
|
|
|
|
def mock_quarantine_db(object_file, server_type):
|
|
self.assertEquals(broker.db_file, object_file)
|
|
self.assertEquals(broker.db_type, server_type)
|
|
|
|
self._patch(patch.object, db_replicator, 'quarantine_db',
|
|
mock_quarantine_db)
|
|
|
|
response = rpc.sync(broker, ('remote_sync', 'hash_', 'id_',
|
|
'created_at', 'put_timestamp',
|
|
'delete_timestamp', 'metadata'))
|
|
|
|
self.assertEquals('404 Not Found', response.status)
|
|
self.assertEquals(404, response.status_int)
|
|
|
|
def test_replicator_sync(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
broker = FakeBroker()
|
|
|
|
response = rpc.sync(broker, (broker.get_sync() + 1, 12345, 'id_',
|
|
'created_at', 'put_timestamp',
|
|
'delete_timestamp',
|
|
'{"meta1": "data1", "meta2": "data2"}'))
|
|
|
|
self.assertEquals({'meta1': 'data1', 'meta2': 'data2'},
|
|
broker.metadata)
|
|
self.assertEquals('created_at', broker.created_at)
|
|
self.assertEquals('put_timestamp', broker.put_timestamp)
|
|
self.assertEquals('delete_timestamp', broker.delete_timestamp)
|
|
|
|
self.assertEquals('200 OK', response.status)
|
|
self.assertEquals(200, response.status_int)
|
|
|
|
def test_rsync_then_merge(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
rpc.rsync_then_merge('sda1', '/srv/swift/blah', ('a', 'b'))
|
|
|
|
def test_merge_items(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
fake_broker = FakeBroker()
|
|
args = ('a', 'b')
|
|
rpc.merge_items(fake_broker, args)
|
|
self.assertEquals(fake_broker.args, args)
|
|
|
|
def test_merge_syncs(self):
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
fake_broker = FakeBroker()
|
|
args = ('a', 'b')
|
|
rpc.merge_syncs(fake_broker, args)
|
|
self.assertEquals(fake_broker.args, (args[0],))
|
|
|
|
def test_complete_rsync_with_bad_input(self):
|
|
drive = '/some/root'
|
|
db_file = __file__
|
|
args = ['old_file']
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
resp = rpc.complete_rsync(drive, db_file, args)
|
|
self.assertTrue(isinstance(resp, HTTPException))
|
|
self.assertEquals(404, resp.status_int)
|
|
resp = rpc.complete_rsync(drive, 'new_db_file', args)
|
|
self.assertTrue(isinstance(resp, HTTPException))
|
|
self.assertEquals(404, resp.status_int)
|
|
|
|
def test_complete_rsync(self):
|
|
drive = mkdtemp()
|
|
args = ['old_file']
|
|
rpc = db_replicator.ReplicatorRpc('/', '/', FakeBroker, False)
|
|
os.mkdir('%s/tmp' % drive)
|
|
old_file = '%s/tmp/old_file' % drive
|
|
new_file = '%s/new_db_file' % drive
|
|
try:
|
|
fp = open(old_file, 'w')
|
|
fp.write('void')
|
|
fp.close
|
|
resp = rpc.complete_rsync(drive, new_file, args)
|
|
self.assertEquals(204, resp.status_int)
|
|
finally:
|
|
rmtree(drive)
|
|
|
|
def test_roundrobin_datadirs(self):
|
|
listdir_calls = []
|
|
isdir_calls = []
|
|
exists_calls = []
|
|
shuffle_calls = []
|
|
|
|
def _listdir(path):
|
|
listdir_calls.append(path)
|
|
if not path.startswith('/srv/node/sda/containers') and \
|
|
not path.startswith('/srv/node/sdb/containers'):
|
|
return []
|
|
path = path[len('/srv/node/sdx/containers'):]
|
|
if path == '':
|
|
return ['123', '456', '789'] # 456 will pretend to be a file
|
|
elif path == '/123':
|
|
return ['abc', 'def.db'] # def.db will pretend to be a file
|
|
elif path == '/123/abc':
|
|
# 11111111111111111111111111111abc will pretend to be a file
|
|
return ['00000000000000000000000000000abc',
|
|
'11111111111111111111111111111abc']
|
|
elif path == '/123/abc/00000000000000000000000000000abc':
|
|
return ['00000000000000000000000000000abc.db',
|
|
# This other.db isn't in the right place, so should be
|
|
# ignored later.
|
|
'000000000000000000000000000other.db',
|
|
'weird1'] # weird1 will pretend to be a dir, if asked
|
|
elif path == '/789':
|
|
return ['ghi', 'jkl'] # jkl will pretend to be a file
|
|
elif path == '/789/ghi':
|
|
# 33333333333333333333333333333ghi will pretend to be a file
|
|
return ['22222222222222222222222222222ghi',
|
|
'33333333333333333333333333333ghi']
|
|
elif path == '/789/ghi/22222222222222222222222222222ghi':
|
|
return ['22222222222222222222222222222ghi.db',
|
|
'weird2'] # weird2 will pretend to be a dir, if asked
|
|
return []
|
|
|
|
def _isdir(path):
|
|
isdir_calls.append(path)
|
|
if not path.startswith('/srv/node/sda/containers') and \
|
|
not path.startswith('/srv/node/sdb/containers'):
|
|
return False
|
|
path = path[len('/srv/node/sdx/containers'):]
|
|
if path in ('/123', '/123/abc',
|
|
'/123/abc/00000000000000000000000000000abc',
|
|
'/123/abc/00000000000000000000000000000abc/weird1',
|
|
'/789', '/789/ghi',
|
|
'/789/ghi/22222222222222222222222222222ghi',
|
|
'/789/ghi/22222222222222222222222222222ghi/weird2'):
|
|
return True
|
|
return False
|
|
|
|
def _exists(arg):
|
|
exists_calls.append(arg)
|
|
return True
|
|
|
|
def _shuffle(arg):
|
|
shuffle_calls.append(arg)
|
|
|
|
orig_listdir = db_replicator.os.listdir
|
|
orig_isdir = db_replicator.os.path.isdir
|
|
orig_exists = db_replicator.os.path.exists
|
|
orig_shuffle = db_replicator.random.shuffle
|
|
try:
|
|
db_replicator.os.listdir = _listdir
|
|
db_replicator.os.path.isdir = _isdir
|
|
db_replicator.os.path.exists = _exists
|
|
db_replicator.random.shuffle = _shuffle
|
|
datadirs = [('/srv/node/sda/containers', 1),
|
|
('/srv/node/sdb/containers', 2)]
|
|
results = list(db_replicator.roundrobin_datadirs(datadirs))
|
|
# The results show that the .db files are returned, the devices
|
|
# interleaved.
|
|
self.assertEquals(results, [
|
|
('123', '/srv/node/sda/containers/123/abc/'
|
|
'00000000000000000000000000000abc/'
|
|
'00000000000000000000000000000abc.db', 1),
|
|
('123', '/srv/node/sdb/containers/123/abc/'
|
|
'00000000000000000000000000000abc/'
|
|
'00000000000000000000000000000abc.db', 2),
|
|
('789', '/srv/node/sda/containers/789/ghi/'
|
|
'22222222222222222222222222222ghi/'
|
|
'22222222222222222222222222222ghi.db', 1),
|
|
('789', '/srv/node/sdb/containers/789/ghi/'
|
|
'22222222222222222222222222222ghi/'
|
|
'22222222222222222222222222222ghi.db', 2)])
|
|
# The listdir calls show that we only listdir the dirs
|
|
self.assertEquals(listdir_calls, [
|
|
'/srv/node/sda/containers',
|
|
'/srv/node/sda/containers/123',
|
|
'/srv/node/sda/containers/123/abc',
|
|
'/srv/node/sdb/containers',
|
|
'/srv/node/sdb/containers/123',
|
|
'/srv/node/sdb/containers/123/abc',
|
|
'/srv/node/sda/containers/789',
|
|
'/srv/node/sda/containers/789/ghi',
|
|
'/srv/node/sdb/containers/789',
|
|
'/srv/node/sdb/containers/789/ghi'])
|
|
# The isdir calls show that we did ask about the things pretending
|
|
# to be files at various levels.
|
|
self.assertEquals(isdir_calls, [
|
|
'/srv/node/sda/containers/123',
|
|
'/srv/node/sda/containers/123/abc',
|
|
('/srv/node/sda/containers/123/abc/'
|
|
'00000000000000000000000000000abc'),
|
|
'/srv/node/sdb/containers/123',
|
|
'/srv/node/sdb/containers/123/abc',
|
|
('/srv/node/sdb/containers/123/abc/'
|
|
'00000000000000000000000000000abc'),
|
|
('/srv/node/sda/containers/123/abc/'
|
|
'11111111111111111111111111111abc'),
|
|
'/srv/node/sda/containers/123/def.db',
|
|
'/srv/node/sda/containers/456',
|
|
'/srv/node/sda/containers/789',
|
|
'/srv/node/sda/containers/789/ghi',
|
|
('/srv/node/sda/containers/789/ghi/'
|
|
'22222222222222222222222222222ghi'),
|
|
('/srv/node/sdb/containers/123/abc/'
|
|
'11111111111111111111111111111abc'),
|
|
'/srv/node/sdb/containers/123/def.db',
|
|
'/srv/node/sdb/containers/456',
|
|
'/srv/node/sdb/containers/789',
|
|
'/srv/node/sdb/containers/789/ghi',
|
|
('/srv/node/sdb/containers/789/ghi/'
|
|
'22222222222222222222222222222ghi'),
|
|
('/srv/node/sda/containers/789/ghi/'
|
|
'33333333333333333333333333333ghi'),
|
|
'/srv/node/sda/containers/789/jkl',
|
|
('/srv/node/sdb/containers/789/ghi/'
|
|
'33333333333333333333333333333ghi'),
|
|
'/srv/node/sdb/containers/789/jkl'])
|
|
# The exists calls are the .db files we looked for as we walked the
|
|
# structure.
|
|
self.assertEquals(exists_calls, [
|
|
('/srv/node/sda/containers/123/abc/'
|
|
'00000000000000000000000000000abc/'
|
|
'00000000000000000000000000000abc.db'),
|
|
('/srv/node/sdb/containers/123/abc/'
|
|
'00000000000000000000000000000abc/'
|
|
'00000000000000000000000000000abc.db'),
|
|
('/srv/node/sda/containers/789/ghi/'
|
|
'22222222222222222222222222222ghi/'
|
|
'22222222222222222222222222222ghi.db'),
|
|
('/srv/node/sdb/containers/789/ghi/'
|
|
'22222222222222222222222222222ghi/'
|
|
'22222222222222222222222222222ghi.db')])
|
|
# Shows that we called shuffle twice, once for each device.
|
|
self.assertEquals(
|
|
shuffle_calls, [['123', '456', '789'], ['123', '456', '789']])
|
|
finally:
|
|
db_replicator.os.listdir = orig_listdir
|
|
db_replicator.os.path.isdir = orig_isdir
|
|
db_replicator.os.path.exists = orig_exists
|
|
db_replicator.random.shuffle = orig_shuffle
|
|
|
|
@mock.patch("swift.common.db_replicator.ReplConnection", mock.Mock())
|
|
def test_http_connect(self):
|
|
node = "node"
|
|
partition = "partition"
|
|
db_file = __file__
|
|
replicator = TestReplicator({})
|
|
replicator._http_connect(node, partition, db_file)
|
|
db_replicator.ReplConnection.assert_has_calls(
|
|
mock.call(node, partition,
|
|
os.path.basename(db_file).split('.', 1)[0],
|
|
replicator.logger))
|
|
|
|
|
|
class TestReplToNode(unittest.TestCase):
|
|
def setUp(self):
|
|
db_replicator.ring = FakeRing()
|
|
self.delete_db_calls = []
|
|
self.broker = FakeBroker()
|
|
self.replicator = TestReplicator({})
|
|
self.fake_node = {'ip': '127.0.0.1', 'device': 'sda1', 'port': 1000}
|
|
self.fake_info = {'id': 'a', 'point': -1, 'max_row': 10, 'hash': 'b',
|
|
'created_at': 100, 'put_timestamp': 0,
|
|
'delete_timestamp': 0, 'count': 0,
|
|
'metadata': {
|
|
'Test': ('Value', normalize_timestamp(1))}}
|
|
self.replicator.logger = mock.Mock()
|
|
self.replicator._rsync_db = mock.Mock(return_value=True)
|
|
self.replicator._usync_db = mock.Mock(return_value=True)
|
|
self.http = ReplHttp('{"id": 3, "point": -1}')
|
|
self.replicator._http_connect = lambda *args: self.http
|
|
|
|
def test_repl_to_node_usync_success(self):
|
|
rinfo = {"id": 3, "point": -1, "max_row": 5, "hash": "c"}
|
|
self.http = ReplHttp(simplejson.dumps(rinfo))
|
|
local_sync = self.broker.get_sync()
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, self.broker, '0', self.fake_info), True)
|
|
self.replicator._usync_db.assert_has_calls([
|
|
mock.call(max(rinfo['point'], local_sync), self.broker,
|
|
self.http, rinfo['id'], self.fake_info['id'])
|
|
])
|
|
|
|
def test_repl_to_node_rsync_success(self):
|
|
rinfo = {"id": 3, "point": -1, "max_row": 4, "hash": "c"}
|
|
self.http = ReplHttp(simplejson.dumps(rinfo))
|
|
self.broker.get_sync()
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, self.broker, '0', self.fake_info), True)
|
|
self.replicator.logger.increment.assert_has_calls([
|
|
mock.call.increment('remote_merges')
|
|
])
|
|
self.replicator._rsync_db.assert_has_calls([
|
|
mock.call(self.broker, self.fake_node, self.http,
|
|
self.fake_info['id'],
|
|
replicate_method='rsync_then_merge',
|
|
replicate_timeout=(self.fake_info['count'] / 2000))
|
|
])
|
|
|
|
def test_repl_to_node_already_in_sync(self):
|
|
rinfo = {"id": 3, "point": -1, "max_row": 10, "hash": "b"}
|
|
self.http = ReplHttp(simplejson.dumps(rinfo))
|
|
self.broker.get_sync()
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, self.broker, '0', self.fake_info), True)
|
|
self.assertEquals(self.replicator._rsync_db.call_count, 0)
|
|
self.assertEquals(self.replicator._usync_db.call_count, 0)
|
|
|
|
def test_repl_to_node_not_found(self):
|
|
self.http = ReplHttp('{"id": 3, "point": -1}', set_status=404)
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, self.broker, '0', self.fake_info), True)
|
|
self.replicator.logger.increment.assert_has_calls([
|
|
mock.call.increment('rsyncs')
|
|
])
|
|
self.replicator._rsync_db.assert_has_calls([
|
|
mock.call(self.broker, self.fake_node, self.http,
|
|
self.fake_info['id'])
|
|
])
|
|
|
|
def test_repl_to_node_drive_not_mounted(self):
|
|
self.http = ReplHttp('{"id": 3, "point": -1}', set_status=507)
|
|
|
|
self.assertRaises(DriveNotMounted, self.replicator._repl_to_node,
|
|
self.fake_node, FakeBroker(), '0', self.fake_info)
|
|
|
|
def test_repl_to_node_300_status(self):
|
|
self.http = ReplHttp('{"id": 3, "point": -1}', set_status=300)
|
|
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, FakeBroker(), '0', self.fake_info), None)
|
|
|
|
def test_repl_to_node_http_connect_fails(self):
|
|
self.replicator._http_connect = lambda *args: None
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, FakeBroker(), '0', self.fake_info), False)
|
|
|
|
def test_repl_to_node_not_response(self):
|
|
self.http = mock.Mock(replicate=mock.Mock(return_value=None))
|
|
self.assertEquals(self.replicator._repl_to_node(
|
|
self.fake_node, FakeBroker(), '0', self.fake_info), False)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|