With this solution, we can catch it cleanly in the standalone nilmdb-server, and test the error paths in our normal test suite.
140 lines
4.7 KiB
Python
140 lines
4.7 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 socket
|
|
import cherrypy
|
|
|
|
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")
|
|
nilmdb.utils.diskusage.du("super-bogus-does-not-exist")
|
|
|
|
def test_cors_allow(self):
|
|
# Just to get some test coverage; these code paths aren't actually
|
|
# used in current code
|
|
cpy = nilmdb.server.serverutil.cherrypy
|
|
(req, resp) = (cpy.request, cpy.response)
|
|
cpy.request.method = "DELETE"
|
|
with assert_raises(cpy.HTTPError):
|
|
nilmdb.server.serverutil.CORS_allow(methods="POST")
|
|
with assert_raises(cpy.HTTPError):
|
|
nilmdb.server.serverutil.CORS_allow(methods=["POST"])
|
|
with assert_raises(cpy.HTTPError):
|
|
nilmdb.server.serverutil.CORS_allow(methods=["GET"])
|
|
with assert_raises(cpy.HTTPError):
|
|
nilmdb.server.serverutil.CORS_allow(methods=[])
|
|
(cpy.request, cpy.response) = (req, resp)
|
|
|
|
def test_cherrypy_failure(self):
|
|
# Test failure of cherrypy to start up because the port is
|
|
# already in use. This also tests the functionality of
|
|
# serverutil:cherrypy_patch_exit()
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
sock.bind(("127.0.0.1", 32180))
|
|
sock.listen(1)
|
|
except OSError:
|
|
raise AssertionError("port 32180 must be free for tests")
|
|
|
|
nilmdb.server.serverutil.cherrypy_patch_exit()
|
|
cherrypy.config.update({
|
|
'environment': 'embedded',
|
|
'server.socket_host': '127.0.0.1',
|
|
'server.socket_port': 32180,
|
|
'engine.autoreload.on': False,
|
|
})
|
|
with assert_raises(Exception) as e:
|
|
cherrypy.engine.start()
|
|
in_("Address already in use", str(e.exception))
|
|
|
|
sock.close()
|