nilmdb/tests/test_nilmdb.py

import nilmdb

from nose.tools import *
from nose.tools import assert_raises
import distutils.version
import json
import itertools 
import os
import shutil
import sys
import cherrypy
import threading
import urllib2
from urllib2 import urlopen, HTTPError
import Queue
import cStringIO

testdb = "tests/testdb"

#@atexit.register
#def cleanup():
#    os.unlink(testdb)

from test_helpers import *

class Test00Nilmdb(object):  # named 00 so it runs first
    def test_NilmDB(self):
        recursive_unlink(testdb)

        with assert_raises(IOError):
            nilmdb.NilmDB("/nonexistant-db/foo")

        db = nilmdb.NilmDB(testdb)
        db.close()
        db = nilmdb.NilmDB(testdb, sync=False)
        db.close()

        # test timer, just to get coverage
        capture = cStringIO.StringIO()
        old = sys.stdout
        sys.stdout = capture
        with nilmdb.Timer("test"):
            nilmdb.timer.time.sleep(0.01)
        sys.stdout = old
        in_("test: ", capture.getvalue())

    def test_stream(self):
        db = nilmdb.NilmDB(testdb, sync=False)
        eq_(db.stream_list(), [])

        # Bad path
        with assert_raises(ValueError):
            db.stream_create("foo/bar/baz", "PrepData")
        with assert_raises(ValueError):
            db.stream_create("/foo", "PrepData")
        # Bad layout type
        with assert_raises(KeyError):
            db.stream_create("/newton/prep", "NoSuchLayout")
        # Bad index columns
        with assert_raises(KeyError):
            db.stream_create("/newton/prep", "PrepData", ["nonexistant"])
        db.stream_create("/newton/prep", "PrepData")
        db.stream_create("/newton/raw", "RawData")
        db.stream_create("/newton/zzz/rawnotch", "RawNotchedData")

        # Verify we got 3 streams
        eq_(db.stream_list(), [ ["/newton/prep", "PrepData"],
                                ["/newton/raw", "RawData"],
                                ["/newton/zzz/rawnotch", "RawNotchedData"]
                                ])
        # Match just one type or one path
        eq_(db.stream_list(layout="RawData"), [ ["/newton/raw", "RawData"] ])
        eq_(db.stream_list(path="/newton/raw"), [ ["/newton/raw", "RawData"] ])

        # Verify that columns were made right
        eq_(len(db.h5file.getNode("/newton/prep").cols), 9)
        eq_(len(db.h5file.getNode("/newton/raw").cols), 7)
        eq_(len(db.h5file.getNode("/newton/zzz/rawnotch").cols), 10)
        assert(db.h5file.getNode("/newton/prep").colindexed["timestamp"])
        assert(not db.h5file.getNode("/newton/prep").colindexed["p1"])

        # Set / get metadata
        eq_(db.stream_get_metadata("/newton/prep"), {})
        eq_(db.stream_get_metadata("/newton/raw"), {})
        meta1 = { "description": "The Data",
                  "v_scale": "1.234" }
        meta2 = { "description": "The Data" }
        meta3 = { "v_scale": "1.234" }
        db.stream_set_metadata("/newton/prep", meta1)
        db.stream_update_metadata("/newton/prep", {})
        db.stream_update_metadata("/newton/raw", meta2)
        db.stream_update_metadata("/newton/raw", meta3)
        eq_(db.stream_get_metadata("/newton/prep"), meta1)
        eq_(db.stream_get_metadata("/newton/raw"), meta1)

        db.close()

class TestBlockingServer(object):
    def setUp(self):
        self.db = nilmdb.NilmDB(testdb, sync=False)

    def tearDown(self):
        self.db.close()
        
    def test_blocking_server(self):
        # Start web app on a custom port
        self.server = nilmdb.Server(self.db, host = "127.0.0.1",
                                    port = 12380, stoppable = True)

        # Run it
        event = threading.Event()
        def run_server():
            self.server.start(blocking = True, event = event)
        thread = threading.Thread(target = run_server)
        thread.start()
        event.wait(timeout = 2)
        
        # Send request to exit.
        req = urlopen("http://127.0.0.1:12380/exit/", timeout = 1)

        # Wait for it
        thread.join()
    
def geturl(path):
    req = urlopen("http://127.0.0.1:12380" + path, timeout = 10)
    return req.read()

def getjson(path):
    return json.loads(geturl(path))

class TestServer(object):

    def setUp(self):
        # Start web app on a custom port
        self.db = nilmdb.NilmDB(testdb, sync=False)
        self.server = nilmdb.Server(self.db, host = "127.0.0.1",
                                    port = 12380, stoppable = False)
        self.server.start(blocking = False)

    def tearDown(self):
        # Close web app
        self.server.stop()
        self.db.close()

    def test_server(self):
        # Make sure we can't force an exit, and test other 404 errors
        for url in [ "/exit", "/", "/favicon.ico" ]:
            with assert_raises(HTTPError) as e:
                geturl(url)
            eq_(e.exception.code, 404)

        # Check version
        eq_(distutils.version.StrictVersion(getjson("/version")),
            distutils.version.StrictVersion(self.server.version))

    def test_stream_list(self):
        # Known streams that got populated by an earlier test (test_nilmdb)
        streams = getjson("/stream/list")

        eq_(streams, [
            ['/newton/prep', 'PrepData'],
            ['/newton/raw', 'RawData'],
            ['/newton/zzz/rawnotch', 'RawNotchedData'],
            ])

        streams = getjson("/stream/list?layout=RawData")
        eq_(streams, [['/newton/raw', 'RawData']])

        streams = getjson("/stream/list?layout=NoSuchLayout")
        eq_(streams, [])


    def test_stream_metadata(self):
        with assert_raises(HTTPError) as e:
            getjson("/stream/get_metadata?path=foo")
        eq_(e.exception.code, 404)

        data = getjson("/stream/get_metadata?path=/newton/prep")
        eq_(data, {'description': 'The Data', 'v_scale': '1.234'})

        data = getjson("/stream/get_metadata?path=/newton/prep"
                       "&key=v_scale")
        eq_(data, {'v_scale': '1.234'})

        data = getjson("/stream/get_metadata?path=/newton/prep"
                       "&key=v_scale&key=description")
        eq_(data, {'description': 'The Data', 'v_scale': '1.234'})

        data = getjson("/stream/get_metadata?path=/newton/prep"
                       "&key=v_scale&key=foo")
        eq_(data, {'foo': None, 'v_scale': '1.234'})

        data = getjson("/stream/get_metadata?path=/newton/prep"
                       "&key=foo")
        eq_(data, {'foo': None})


    def test_insert(self):
        # GET instead of POST (no body)
        # (actual POST test is done by client code)
        with assert_raises(HTTPError) as e:
            getjson("/stream/insert?path=/newton/prep")
        eq_(e.exception.code, 400)