Add unit test for account/auditor.py

* Here lack tests for a long time; let's add tests for it.
* use mock module instead of monkeypatch in both account/test_auditor.py
  and container/test_auditor.py

Change-Id: I9ea23dc5034f7147629db59b1ef6cb491de9049f
This commit is contained in:
Kun Huang 2013-08-04 17:31:14 +08:00 committed by Kun Huang
parent 8a8499805b
commit 6fb327bc97
2 changed files with 108 additions and 13 deletions

View File

@ -13,15 +13,109 @@
# See the License for the specific language governing permissions and
# limitations under the License.
# TODO: Tests
import unittest
import mock
import time
import os
import random
from tempfile import mkdtemp
from shutil import rmtree
from swift.account import auditor
from test.unit import FakeLogger
class FakeAccountBroker(object):
def __init__(self, path):
self.path = path
self.db_file = path
self.file = os.path.basename(path)
def is_deleted(self):
return False
def get_info(self):
if self.file.startswith('fail'):
raise ValueError
if self.file.startswith('true'):
return 'ok'
class TestAuditor(unittest.TestCase):
def test_placeholder(self):
pass
def setUp(self):
self.testdir = os.path.join(mkdtemp(), 'tmp_test_account_auditor')
self.logger = FakeLogger()
rmtree(self.testdir, ignore_errors=1)
os.mkdir(self.testdir)
fnames = ['true1.db', 'true2.db', 'true3.db',
'fail1.db', 'fail2.db']
for fn in fnames:
with open(os.path.join(self.testdir, fn), 'w+') as f:
f.write(' ')
def tearDown(self):
rmtree(os.path.dirname(self.testdir), ignore_errors=1)
@mock.patch('swift.account.auditor.AccountBroker', FakeAccountBroker)
def test_run_forever(self):
sleep_times = random.randint(5, 10)
call_times = sleep_times - 1
class FakeTime(object):
def __init__(self):
self.times = 0
def sleep(self, sec):
self.times += 1
if self.times < sleep_times:
time.sleep(0.1)
else:
# stop forever by an error
raise ValueError()
def time(self):
return time.time()
conf = {}
test_auditor = auditor.AccountAuditor(conf)
with mock.patch('swift.account.auditor.time', FakeTime()):
def fake_audit_location_generator(*args, **kwargs):
files = os.listdir(self.testdir)
return [(os.path.join(self.testdir, f), '', '') for f in files]
auditor.audit_location_generator = fake_audit_location_generator
self.assertRaises(ValueError, test_auditor.run_forever)
self.assertEquals(test_auditor.account_failures, 2 * call_times)
self.assertEquals(test_auditor.account_passes, 3 * call_times)
@mock.patch('swift.account.auditor.AccountBroker', FakeAccountBroker)
def test_run_once(self):
conf = {}
test_auditor = auditor.AccountAuditor(conf)
def fake_audit_location_generator(*args, **kwargs):
files = os.listdir(self.testdir)
return [(os.path.join(self.testdir, f), '', '') for f in files]
auditor.audit_location_generator = fake_audit_location_generator
test_auditor.run_once()
self.assertEquals(test_auditor.account_failures, 2)
self.assertEquals(test_auditor.account_passes, 3)
@mock.patch('swift.account.auditor.AccountBroker', FakeAccountBroker)
def test_account_auditor(self):
conf = {}
test_auditor = auditor.AccountAuditor(conf)
files = os.listdir(self.testdir)
for f in files:
path = os.path.join(self.testdir, f)
test_auditor.account_audit(path)
self.assertEquals(test_auditor.account_failures, 2)
self.assertEquals(test_auditor.account_passes, 3)
if __name__ == '__main__':
unittest.main()

View File

@ -14,6 +14,7 @@
# limitations under the License.
import unittest
import mock
import time
import os
import random
@ -56,6 +57,7 @@ class TestAuditor(unittest.TestCase):
def tearDown(self):
rmtree(os.path.dirname(self.testdir), ignore_errors=1)
@mock.patch('swift.container.auditor.ContainerBroker', FakeContainerBroker)
def test_run_forever(self):
sleep_times = random.randint(5, 10)
call_times = sleep_times - 1
@ -77,23 +79,22 @@ class TestAuditor(unittest.TestCase):
conf = {}
test_auditor = auditor.ContainerAuditor(conf)
auditor.ContainerBroker = FakeContainerBroker
auditor.time = FakeTime()
def fake_audit_location_generator(*args, **kwargs):
files = os.listdir(self.testdir)
return [(os.path.join(self.testdir, f), '', '') for f in files]
with mock.patch('swift.container.auditor.time', FakeTime()):
def fake_audit_location_generator(*args, **kwargs):
files = os.listdir(self.testdir)
return [(os.path.join(self.testdir, f), '', '') for f in files]
auditor.audit_location_generator = fake_audit_location_generator
auditor.audit_location_generator = fake_audit_location_generator
self.assertRaises(ValueError, test_auditor.run_forever)
self.assertRaises(ValueError, test_auditor.run_forever)
self.assertEquals(test_auditor.container_failures, 2 * call_times)
self.assertEquals(test_auditor.container_passes, 3 * call_times)
@mock.patch('swift.container.auditor.ContainerBroker', FakeContainerBroker)
def test_run_once(self):
conf = {}
test_auditor = auditor.ContainerAuditor(conf)
auditor.ContainerBroker = FakeContainerBroker
def fake_audit_location_generator(*args, **kwargs):
files = os.listdir(self.testdir)
@ -105,10 +106,10 @@ class TestAuditor(unittest.TestCase):
self.assertEquals(test_auditor.container_failures, 2)
self.assertEquals(test_auditor.container_passes, 3)
@mock.patch('swift.container.auditor.ContainerBroker', FakeContainerBroker)
def test_container_auditor(self):
conf = {}
test_auditor = auditor.ContainerAuditor(conf)
auditor.ContainerBroker = FakeContainerBroker
files = os.listdir(self.testdir)
for f in files:
path = os.path.join(self.testdir, f)