本文整理汇总了Python中swift.obj.server.DiskFile类的典型用法代码示例。如果您正苦于以下问题：Python DiskFile类的具体用法？Python DiskFile怎么用？Python DiskFile使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了DiskFile类的14个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setup_bad_zero_byte
def setup_bad_zero_byte(self, with_ts=False):
self.auditor = auditor.ObjectAuditor(self.conf)
self.auditor.log_time = 0
cur_part = '0'
ts_file_path = ''
if with_ts:
name_hash = hash_path('a', 'c', 'o')
dir_path = os.path.join(self.devices, 'sda',
storage_directory(DATADIR, cur_part, name_hash))
ts_file_path = os.path.join(dir_path, '99999.ts')
if not os.path.exists(dir_path):
mkdirs(dir_path)
fp = open(ts_file_path, 'w')
fp.close()
disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
etag = md5()
with disk_file.mkstemp() as (fd, tmppath):
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': str(normalize_timestamp(time.time())),
'Content-Length': 10,
}
disk_file.put(fd, tmppath, metadata)
etag = md5()
etag = etag.hexdigest()
metadata['ETag'] = etag
write_metadata(fd, metadata)
if disk_file.data_file:
return disk_file.data_file
return ts_file_path
开发者ID:pombredanne，项目名称:swift，代码行数:34，代码来源:test_auditor.py
示例2: test_object_audit_diff_data
def test_object_audit_diff_data(self):
self.auditor = auditor.AuditorWorker(self.conf)
cur_part = '0'
disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
data = '0' * 1024
etag = md5()
timestamp = str(normalize_timestamp(time.time()))
with disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(fd).st_size),
}
disk_file.put(fd, tmppath, metadata)
pre_quarantines = self.auditor.quarantines
self.auditor.object_audit(
os.path.join(disk_file.datadir, timestamp + '.data'),
'sda', cur_part)
self.assertEquals(self.auditor.quarantines, pre_quarantines)
etag = md5()
etag.update('1' + '0' * 1023)
etag = etag.hexdigest()
metadata['ETag'] = etag
write_metadata(fd, metadata)
self.auditor.object_audit(
os.path.join(disk_file.datadir, timestamp + '.data'),
'sda', cur_part)
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
开发者ID:pombredanne，项目名称:swift，代码行数:33，代码来源:test_auditor.py
示例3: test_object_run_fast_track_non_zero
def test_object_run_fast_track_non_zero(self):
self.auditor = auditor.ObjectAuditor(self.conf)
self.auditor.log_time = 0
cur_part = '0'
disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
data = '0' * 1024
etag = md5()
with disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': str(normalize_timestamp(time.time())),
'Content-Length': str(os.fstat(fd).st_size),
}
disk_file.put(fd, tmppath, metadata)
etag = md5()
etag.update('1' + '0' * 1023)
etag = etag.hexdigest()
metadata['ETag'] = etag
write_metadata(fd, metadata)
quarantine_path = os.path.join(self.devices,
'sda', 'quarantined', 'objects')
self.auditor.run_once(zero_byte_fps=50)
self.assertFalse(os.path.isdir(quarantine_path))
self.auditor.run_once()
self.assertTrue(os.path.isdir(quarantine_path))
开发者ID:pombredanne，项目名称:swift，代码行数:29，代码来源:test_auditor.py
示例4: test_object_audit_diff_data
def test_object_audit_diff_data(self):
self.auditor = auditor.AuditorWorker(self.conf)
data = "0" * 1024
etag = md5()
timestamp = str(normalize_timestamp(time.time()))
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {"ETag": etag, "X-Timestamp": timestamp, "Content-Length": str(os.fstat(fd).st_size)}
self.disk_file.put(fd, tmppath, metadata)
pre_quarantines = self.auditor.quarantines
# remake so it will have metadata
self.disk_file = DiskFile(self.devices, "sda", "0", "a", "c", "o", self.logger)
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines)
etag = md5()
etag.update("1" + "0" * 1023)
etag = etag.hexdigest()
metadata["ETag"] = etag
write_metadata(fd, metadata)
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
开发者ID:jness，项目名称:python-swift，代码行数:25，代码来源:test_auditor.py
示例5: test_object_run_once_multi_devices
def test_object_run_once_multi_devices(self):
self.auditor = auditor.AuditorWorker(self.conf, self.logger)
timestamp = str(normalize_timestamp(time.time()))
pre_quarantines = self.auditor.quarantines
data = '0' * 10
etag = md5()
with self.disk_file.writer() as writer:
writer.write(data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(writer.fd).st_size),
}
writer.put(metadata)
self.auditor.audit_all_objects()
self.disk_file = DiskFile(self.devices, 'sdb', '0', 'a', 'c',
'ob', self.logger)
data = '1' * 10
etag = md5()
with self.disk_file.writer() as writer:
writer.write(data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(writer.fd).st_size),
}
writer.put(metadata)
os.write(writer.fd, 'extra_data')
self.auditor.audit_all_objects()
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
开发者ID:ngasiproj，项目名称:swift，代码行数:34，代码来源:test_auditor.py
示例6: test_object_run_once_multi_devices
def test_object_run_once_multi_devices(self):
self.auditor = auditor.AuditorWorker(self.conf)
timestamp = str(normalize_timestamp(time.time()))
pre_quarantines = self.auditor.quarantines
data = "0" * 10
etag = md5()
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {"ETag": etag, "X-Timestamp": timestamp, "Content-Length": str(os.fstat(fd).st_size)}
self.disk_file.put(fd, tmppath, metadata)
self.disk_file.close()
self.auditor.audit_all_objects()
self.disk_file = DiskFile(self.devices, "sdb", "0", "a", "c", "ob", self.logger)
data = "1" * 10
etag = md5()
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {"ETag": etag, "X-Timestamp": timestamp, "Content-Length": str(os.fstat(fd).st_size)}
self.disk_file.put(fd, tmppath, metadata)
self.disk_file.close()
os.write(fd, "extra_data")
self.auditor.audit_all_objects()
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
开发者ID:jness，项目名称:python-swift，代码行数:27，代码来源:test_auditor.py
示例7: test_object_audit_diff_data
def test_object_audit_diff_data(self):
self.auditor = auditor.AuditorWorker(self.conf, self.logger)
data = '0' * 1024
etag = md5()
timestamp = str(normalize_timestamp(time.time()))
with self.disk_file.mkstemp() as fd:
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(fd).st_size),
'Original-Content-Length': str(os.fstat(fd).st_size)
}
self.disk_file.put(fd, metadata)
pre_quarantines = self.auditor.quarantines
# remake so it will have metadata
self.disk_file = DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o',
self.logger)
self.auditor.object_audit(
os.path.join(self.disk_file.datadir, timestamp + '.data'),
'sda', '0')
self.assertEquals(self.auditor.quarantines, pre_quarantines)
etag = md5()
etag.update('1' + '0' * 1023)
etag = etag.hexdigest()
metadata['ETag'] = etag
write_metadata(fd, metadata)
self.auditor.object_audit(
os.path.join(self.disk_file.datadir, timestamp + '.data'),
'sda', '0')
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
开发者ID:Manasa7，项目名称:swift-encrypt，代码行数:35，代码来源:test_auditor.py
示例8: setUp
def setUp(self):
self.testdir = os.path.join(mkdtemp(), "tmp_test_object_auditor")
self.devices = os.path.join(self.testdir, "node")
self.logger = FakeLogger()
rmtree(self.testdir, ignore_errors=1)
mkdirs(os.path.join(self.devices, "sda"))
self.objects = os.path.join(self.devices, "sda", "objects")
os.mkdir(os.path.join(self.devices, "sdb"))
self.objects_2 = os.path.join(self.devices, "sdb", "objects")
os.mkdir(self.objects)
self.parts = {}
for part in ["0", "1", "2", "3"]:
self.parts[part] = os.path.join(self.objects, part)
os.mkdir(os.path.join(self.objects, part))
self.conf = dict(devices=self.devices, mount_check="false")
self.disk_file = DiskFile(self.devices, "sda", "0", "a", "c", "o", self.logger)
开发者ID:jness，项目名称:python-swift，代码行数:19，代码来源:test_auditor.py
示例9: setUp
def setUp(self):
self.testdir = os.path.join(mkdtemp(), 'tmp_test_object_auditor')
self.devices = os.path.join(self.testdir, 'node')
self.logger = FakeLogger()
rmtree(self.testdir, ignore_errors=1)
mkdirs(os.path.join(self.devices, 'sda'))
self.objects = os.path.join(self.devices, 'sda', 'objects')
os.mkdir(os.path.join(self.devices, 'sdb'))
self.objects_2 = os.path.join(self.devices, 'sdb', 'objects')
os.mkdir(self.objects)
self.parts = {}
for part in ['0', '1', '2', '3']:
self.parts[part] = os.path.join(self.objects, part)
os.mkdir(os.path.join(self.objects, part))
self.conf = dict(
devices=self.devices,
mount_check='false')
self.disk_file = DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o',
self.logger)
开发者ID:BillTheBest，项目名称:swift，代码行数:22，代码来源:test_auditor.py
示例10: setUp
def setUp(self):
self.testdir = os.path.join(mkdtemp(), 'tmp_test_object_auditor')
self.devices = os.path.join(self.testdir, 'node')
self.logger = FakeLogger()
rmtree(self.testdir, ignore_errors=1)
mkdirs(os.path.join(self.devices, 'sda'))
self.objects = os.path.join(self.devices, 'sda', 'objects')
os.mkdir(os.path.join(self.devices, 'sdb'))
self.objects_2 = os.path.join(self.devices, 'sdb', 'objects')
os.mkdir(self.objects)
self.parts = {}
for part in ['0', '1', '2', '3']:
self.parts[part] = os.path.join(self.objects, part)
os.mkdir(os.path.join(self.objects, part))
self.conf = dict(
devices=self.devices,
mount_check='false')
self.disk_file = DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o',
self.logger)
try:
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd,data)
etag.update(data)
timestamp = str(normalize_timestamp(time.time()))
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(fd).st_size)
}
self.disk_file.put(fd, tmppath, metadata)
except:
raise SkipTest
开发者ID:lixmgl，项目名称:Intern_OpenStack_Swift，代码行数:36，代码来源:test_auditor.py
示例11: test_object_run_once_no_sda
def test_object_run_once_no_sda(self):
self.auditor = auditor.AuditorWorker(self.conf)
cur_part = '0'
timestamp = str(normalize_timestamp(time.time()))
pre_quarantines = self.auditor.quarantines
disk_file = DiskFile(self.devices, 'sdb', cur_part, 'a', 'c', 'o')
data = '0' * 1024
etag = md5()
with disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(fd).st_size),
}
disk_file.put(fd, tmppath, metadata)
disk_file.close()
os.write(fd, 'extra_data')
self.auditor.audit_all_objects()
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
开发者ID:pombredanne，项目名称:swift，代码行数:22，代码来源:test_auditor.py
示例12: TestAuditor
class TestAuditor(unittest.TestCase):
def setUp(self):
self.testdir = os.path.join(mkdtemp(), 'tmp_test_object_auditor')
self.devices = os.path.join(self.testdir, 'node')
self.logger = FakeLogger()
rmtree(self.testdir, ignore_errors=1)
mkdirs(os.path.join(self.devices, 'sda'))
self.objects = os.path.join(self.devices, 'sda', 'objects')
os.mkdir(os.path.join(self.devices, 'sdb'))
self.objects_2 = os.path.join(self.devices, 'sdb', 'objects')
os.mkdir(self.objects)
self.parts = {}
for part in ['0', '1', '2', '3']:
self.parts[part] = os.path.join(self.objects, part)
os.mkdir(os.path.join(self.objects, part))
self.conf = dict(
devices=self.devices,
mount_check='false')
self.disk_file = DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o',
self.logger)
def tearDown(self):
rmtree(os.path.dirname(self.testdir), ignore_errors=1)
unit.xattr_data = {}
def test_object_audit_extra_data(self):
self.auditor = auditor.AuditorWorker(self.conf)
data = '0' * 1024
etag = md5()
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
timestamp = str(normalize_timestamp(time.time()))
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(fd).st_size),
}
self.disk_file.put(fd, tmppath, metadata)
pre_quarantines = self.auditor.quarantines
self.auditor.object_audit(
os.path.join(self.disk_file.datadir, timestamp + '.data'),
'sda', '0')
self.assertEquals(self.auditor.quarantines, pre_quarantines)
os.write(fd, 'extra_data')
self.auditor.object_audit(
os.path.join(self.disk_file.datadir, timestamp + '.data'),
'sda', '0')
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
def test_object_audit_diff_data(self):
self.auditor = auditor.AuditorWorker(self.conf)
data = '0' * 1024
etag = md5()
timestamp = str(normalize_timestamp(time.time()))
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {
'ETag': etag,
'X-Timestamp': timestamp,
'Content-Length': str(os.fstat(fd).st_size),
}
self.disk_file.put(fd, tmppath, metadata)
pre_quarantines = self.auditor.quarantines
# remake so it will have metadata
self.disk_file = DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o',
self.logger)
self.auditor.object_audit(
os.path.join(self.disk_file.datadir, timestamp + '.data'),
'sda', '0')
self.assertEquals(self.auditor.quarantines, pre_quarantines)
etag = md5()
etag.update('1' + '0' * 1023)
etag = etag.hexdigest()
metadata['ETag'] = etag
write_metadata(fd, metadata)
self.auditor.object_audit(
os.path.join(self.disk_file.datadir, timestamp + '.data'),
'sda', '0')
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
def test_object_audit_no_meta(self):
timestamp = str(normalize_timestamp(time.time()))
path = os.path.join(self.disk_file.datadir, timestamp + '.data')
mkdirs(self.disk_file.datadir)
fp = open(path, 'w')
fp.write('0' * 1024)
fp.close()
invalidate_hash(os.path.dirname(self.disk_file.datadir))
#.........这里部分代码省略.........
开发者ID:BillTheBest，项目名称:swift，代码行数:101，代码来源:test_auditor.py
示例13: close
def close(self, verify_file=True):
rat[0] = True
DiskFile.close(self, verify_file=verify_file)
开发者ID:BillTheBest，项目名称:swift，代码行数:3，代码来源:test_auditor.py
示例14: TestAuditor
class TestAuditor(unittest.TestCase):
def setUp(self):
self.testdir = os.path.join(mkdtemp(), "tmp_test_object_auditor")
self.devices = os.path.join(self.testdir, "node")
self.logger = FakeLogger()
rmtree(self.testdir, ignore_errors=1)
mkdirs(os.path.join(self.devices, "sda"))
self.objects = os.path.join(self.devices, "sda", "objects")
os.mkdir(os.path.join(self.devices, "sdb"))
self.objects_2 = os.path.join(self.devices, "sdb", "objects")
os.mkdir(self.objects)
self.parts = {}
for part in ["0", "1", "2", "3"]:
self.parts[part] = os.path.join(self.objects, part)
os.mkdir(os.path.join(self.objects, part))
self.conf = dict(devices=self.devices, mount_check="false")
self.disk_file = DiskFile(self.devices, "sda", "0", "a", "c", "o", self.logger)
def tearDown(self):
rmtree(os.path.dirname(self.testdir), ignore_errors=1)
unit.xattr_data = {}
def test_object_audit_extra_data(self):
self.auditor = auditor.AuditorWorker(self.conf)
data = "0" * 1024
etag = md5()
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
timestamp = str(normalize_timestamp(time.time()))
metadata = {"ETag": etag, "X-Timestamp": timestamp, "Content-Length": str(os.fstat(fd).st_size)}
self.disk_file.put(fd, tmppath, metadata)
pre_quarantines = self.auditor.quarantines
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines)
os.write(fd, "extra_data")
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
def test_object_audit_diff_data(self):
self.auditor = auditor.AuditorWorker(self.conf)
data = "0" * 1024
etag = md5()
timestamp = str(normalize_timestamp(time.time()))
with self.disk_file.mkstemp() as (fd, tmppath):
os.write(fd, data)
etag.update(data)
etag = etag.hexdigest()
metadata = {"ETag": etag, "X-Timestamp": timestamp, "Content-Length": str(os.fstat(fd).st_size)}
self.disk_file.put(fd, tmppath, metadata)
pre_quarantines = self.auditor.quarantines
# remake so it will have metadata
self.disk_file = DiskFile(self.devices, "sda", "0", "a", "c", "o", self.logger)
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines)
etag = md5()
etag.update("1" + "0" * 1023)
etag = etag.hexdigest()
metadata["ETag"] = etag
write_metadata(fd, metadata)
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
def test_object_audit_no_meta(self):
timestamp = str(normalize_timestamp(time.time()))
path = os.path.join(self.disk_file.datadir, timestamp + ".data")
mkdirs(self.disk_file.datadir)
fp = open(path, "w")
fp.write("0" * 1024)
fp.close()
invalidate_hash(os.path.dirname(self.disk_file.datadir))
self.auditor = auditor.AuditorWorker(self.conf)
pre_quarantines = self.auditor.quarantines
self.auditor.object_audit(os.path.join(self.disk_file.datadir, timestamp + ".data"), "sda", "0")
self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
def test_object_audit_bad_args(self):
self.auditor = auditor.AuditorWorker(self.conf)
pre_errors = self.auditor.errors
self.auditor.object_audit(5, "sda", "0")
self.assertEquals(self.auditor.errors, pre_errors + 1)
pre_errors = self.auditor.errors
self.auditor.object_audit("badpath", "sda", "0")
self.assertEquals(self.auditor.errors, pre_errors) # just returns
def test_object_run_once_pass(self):
self.auditor = auditor.AuditorWorker(self.conf)
self.auditor.log_time = 0
timestamp = str(normalize_timestamp(time.time()))
pre_quarantines = self.auditor.quarantines
data = "0" * 1024
etag = md5()
#.........这里部分代码省略.........
开发者ID:jness，项目名称:python-swift，代码行数:101，代码来源:test_auditor.py
注：本文中的swift.obj.server.DiskFile类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论