nilmdb/tests/test_misc.py

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()