def test_read_disjointed_changing_total_space(self):
"""
Total space measurements are only sent when (a) none have ever
been sent, or (b) the value has changed since the last time
data was collected. This test ensures that the (b) criteria
is checked per-mount point. The sample statvfs() function
only provides changing total space for /; therefore, new
messages should only be queued for / after the first message
is created.
"""
counter = mock_counter(1)
def statvfs(path, multiplier=lambda: next(counter)):
if path == "/":
return os.statvfs_result(
(4096, 0, mb(1000), mb(100), 0, 0, 0, 0, 0, 0))
return os.statvfs_result(
(4096, 0, mb(multiplier() * 1000), mb(100), 0, 0, 0, 0, 0, 0))
filename = self.makeFile("""\
/dev/hda1 / ext3 rw 0 0
/dev/hde1 /mnt/hde1 ext3 rw 0 0
""")
plugin = self.get_mount_info(mounts_file=filename, statvfs=statvfs,
create_time=self.reactor.time,
interval=self.monitor.step_size,
mtab_file=filename)
self.monitor.add(plugin)
self.reactor.advance(self.monitor.step_size * 2)
message = plugin.create_mount_info_message()
self.assertTrue(message)
mount_info = message.get("mount-info", ())
self.assertEqual(len(mount_info), 3)
self.assertEqual(mount_info[0][0], self.monitor.step_size)
self.assertEqual(mount_info[0][1],
{"device": "/dev/hda1", "mount-point": "/",
"filesystem": "ext3", "total-space": 4096000})
self.assertEqual(mount_info[1][0], self.monitor.step_size)
self.assertEqual(mount_info[1][1],
{"device": "/dev/hde1", "mount-point": "/mnt/hde1",
"filesystem": "ext3", "total-space": 4096000})
self.assertEqual(mount_info[2][0], self.monitor.step_size * 2)
self.assertEqual(mount_info[2][1],
{"device": "/dev/hde1", "mount-point": "/mnt/hde1",
"filesystem": "ext3", "total-space": 8192000})
评论列表
文章目录
