96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
from nose.tools import *
|
|
from nose.tools import assert_raises
|
|
from testutil.helpers import *
|
|
|
|
import io
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
import nilmdb.server
|
|
from nilmdb.utils import timer, lock
|
|
|
|
class TestMisc(object):
|
|
def test_timer(self):
|
|
capture = io.StringIO()
|
|
old = sys.stdout
|
|
sys.stdout = capture
|
|
with nilmdb.utils.Timer("test"):
|
|
time.sleep(0.01)
|
|
with nilmdb.utils.Timer("test syslog", tosyslog=True):
|
|
time.sleep(0.01)
|
|
sys.stdout = old
|
|
in_("test: ", capture.getvalue())
|
|
|
|
def test_lock(self):
|
|
with open("/dev/null") as f:
|
|
eq_(nilmdb.utils.lock.exclusive_lock(f), True)
|
|
nilmdb.utils.lock.exclusive_unlock(f)
|
|
# Test error conditions
|
|
class FakeFile():
|
|
def __init__(self, fileno):
|
|
self._fileno = fileno
|
|
def fileno(self):
|
|
return self._fileno
|
|
with assert_raises(TypeError):
|
|
nilmdb.utils.lock.exclusive_lock(FakeFile('none'))
|
|
with assert_raises(ValueError):
|
|
nilmdb.utils.lock.exclusive_lock(FakeFile(-1))
|
|
with assert_raises(IOError):
|
|
nilmdb.utils.lock.exclusive_lock(FakeFile(12345))
|
|
|
|
# Lock failure is tested in test_bulkdata
|
|
|
|
def test_replace_file(self):
|
|
fn = b"tests/misc-testdb/file"
|
|
try:
|
|
os.mkdir(os.path.dirname(fn))
|
|
except FileExistsError:
|
|
pass
|
|
with open(fn, "wb") as f:
|
|
f.write(b"hello, world")
|
|
nilmdb.utils.atomic.replace_file(fn, b"goodbye, world")
|
|
with open(fn, "rb") as f:
|
|
eq_(f.read(), b"goodbye, world")
|
|
|
|
def test_punch(self):
|
|
fn = b"tests/misc-testdb/punchit"
|
|
try:
|
|
os.mkdir(os.path.dirname(fn))
|
|
except FileExistsError:
|
|
pass
|
|
with open(fn, "wb") as f:
|
|
f.write(b"hello, world")
|
|
nilmdb.utils.fallocate.punch_hole(fn, 3, 5)
|
|
with open(fn, "rb") as f:
|
|
eq_(f.read(), b"hel\0\0\0\0\0orld")
|
|
with assert_raises(OSError):
|
|
nilmdb.utils.fallocate.punch_hole(fn, 1, -1, False)
|
|
with assert_raises(OSError):
|
|
nilmdb.utils.fallocate.punch_hole("/", 1, 1, False)
|
|
# no exception because we ignore errors by default
|
|
nilmdb.utils.fallocate.punch_hole(fn, 1, -1)
|
|
|
|
def test_diskusage(self):
|
|
hs = nilmdb.utils.diskusage.human_size
|
|
eq_(hs(0), "0 bytes")
|
|
eq_(hs(1), "1 byte")
|
|
eq_(hs(1023), "1023 bytes")
|
|
|
|
eq_(hs(1024), "1 kiB")
|
|
|
|
eq_(hs(1048575), "1024 kiB")
|
|
eq_(hs(1048576), "1.0 MiB")
|
|
|
|
eq_(hs(1073741823), "1024.0 MiB")
|
|
eq_(hs(1073741824), "1.00 GiB")
|
|
|
|
eq_(hs(1099511627775), "1024.00 GiB")
|
|
eq_(hs(1099511627776), "1.00 TiB")
|
|
|
|
eq_(hs(1099511627776 * 5000.1234), "5000.12 TiB")
|
|
|
|
nilmdb.utils.diskusage.du("/dev")
|
|
with assert_raises(OSError):
|
|
nilmdb.utils.diskusage.du("/dev/null/bogus")
|