import nilmdb from nose.tools import * from nose.tools import assert_raises import distutils.version import simplejson as json import itertools import os import shutil import sys import cherrypy import threading import urllib2 from urllib2 import urlopen, HTTPError import Queue import cStringIO import time 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.utils.Timer("test"): 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(ValueError): db.stream_create("/newton/prep", "NoSuchLayout") 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 (pytables specific) if "h5file" in db.data.__dict__: h5file = db.data.h5file eq_(len(h5file.getNode("/newton/prep").cols), 9) eq_(len(h5file.getNode("/newton/raw").cols), 7) eq_(len(h5file.getNode("/newton/zzz/rawnotch").cols), 10) assert(not h5file.getNode("/newton/prep").colindexed["timestamp"]) assert(not h5file.getNode("/newton/prep").colindexed["c1"]) # 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&start=0&end=0") eq_(e.exception.code, 400)