|
|
@@ -1,11 +1,15 @@ |
|
|
|
import nilmdb |
|
|
|
from nilmdb import NilmDB |
|
|
|
|
|
|
|
from nose.tools import * |
|
|
|
from nose.tools import assert_raises |
|
|
|
from distutils.version import StrictVersion as V |
|
|
|
import json |
|
|
|
import itertools |
|
|
|
import os |
|
|
|
import sys |
|
|
|
import atexit |
|
|
|
import cherrypy |
|
|
|
import threading |
|
|
|
import urllib2 |
|
|
|
|
|
|
|
testdb = "tests/test.db" |
|
|
|
|
|
|
@@ -13,41 +17,97 @@ testdb = "tests/test.db" |
|
|
|
#def cleanup(): |
|
|
|
# os.unlink(testdb) |
|
|
|
|
|
|
|
def test_NilmDB(): |
|
|
|
try: |
|
|
|
os.unlink(testdb) |
|
|
|
except: |
|
|
|
pass |
|
|
|
|
|
|
|
with assert_raises(IOError): |
|
|
|
nilmdb.NilmDB("/nonexistant-db/foo") |
|
|
|
|
|
|
|
db = nilmdb.NilmDB(testdb) |
|
|
|
db.close() |
|
|
|
db = nilmdb.NilmDB(testdb) |
|
|
|
|
|
|
|
def test_stream(): |
|
|
|
db = nilmdb.NilmDB(testdb) |
|
|
|
assert(db.stream_list() == []) |
|
|
|
|
|
|
|
# Bad path |
|
|
|
with assert_raises(ValueError): |
|
|
|
db.stream_create("/foo", nilmdb.layout.PrepData) |
|
|
|
# Bad index columns |
|
|
|
with assert_raises(KeyError): |
|
|
|
db.stream_create("/newton/prep", nilmdb.layout.PrepData, ["nonexistant"]) |
|
|
|
db.stream_create("/newton/prep", nilmdb.layout.PrepData) |
|
|
|
db.stream_create("/newton/raw", nilmdb.layout.RawData) |
|
|
|
db.stream_create("/newton/zzz/rawnotch", nilmdb.layout.RawNotchedData) |
|
|
|
|
|
|
|
# Verify we got 3 streams |
|
|
|
assert(db.stream_list() == [ "/newton/prep", |
|
|
|
"/newton/raw", |
|
|
|
"/newton/zzz/rawnotch" ]) |
|
|
|
|
|
|
|
# Verify that columns were made right |
|
|
|
assert(len(db.h5file.getNode("/newton/prep").cols) == 9) |
|
|
|
assert(len(db.h5file.getNode("/newton/raw").cols) == 7) |
|
|
|
assert(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"]) |
|
|
|
class TestNilmdb(object): |
|
|
|
def test_NilmDB(self): |
|
|
|
try: |
|
|
|
os.unlink(testdb) |
|
|
|
except: |
|
|
|
pass |
|
|
|
|
|
|
|
with assert_raises(IOError): |
|
|
|
nilmdb.NilmDB("/nonexistant-db/foo") |
|
|
|
|
|
|
|
db = nilmdb.NilmDB(testdb) |
|
|
|
db.close() |
|
|
|
db = nilmdb.NilmDB(testdb) |
|
|
|
db.close() |
|
|
|
|
|
|
|
def test_stream(self): |
|
|
|
db = nilmdb.NilmDB(testdb) |
|
|
|
assert(db.stream_list() == []) |
|
|
|
|
|
|
|
# Bad path |
|
|
|
with assert_raises(ValueError): |
|
|
|
db.stream_create("/foo", nilmdb.layout.PrepData) |
|
|
|
# Bad index columns |
|
|
|
with assert_raises(KeyError): |
|
|
|
db.stream_create("/newton/prep", nilmdb.layout.PrepData, ["nonexistant"]) |
|
|
|
db.stream_create("/newton/prep", nilmdb.layout.PrepData) |
|
|
|
db.stream_create("/newton/raw", nilmdb.layout.RawData) |
|
|
|
db.stream_create("/newton/zzz/rawnotch", nilmdb.layout.RawNotchedData) |
|
|
|
|
|
|
|
# Verify we got 3 streams |
|
|
|
assert(db.stream_list() == [ "/newton/prep", |
|
|
|
"/newton/raw", |
|
|
|
"/newton/zzz/rawnotch" ]) |
|
|
|
|
|
|
|
# Verify that columns were made right |
|
|
|
assert(len(db.h5file.getNode("/newton/prep").cols) == 9) |
|
|
|
assert(len(db.h5file.getNode("/newton/raw").cols) == 7) |
|
|
|
assert(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"]) |
|
|
|
db.close() |
|
|
|
|
|
|
|
class TestBlockingServer(object): |
|
|
|
def test_blocking_server(self): |
|
|
|
# Start web app on a custom port |
|
|
|
self.server = nilmdb.Server(testdb, 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() |
|
|
|
|
|
|
|
# Send request to exit. |
|
|
|
req = urllib2.urlopen("http://127.0.0.1:12380/exit/") |
|
|
|
|
|
|
|
# Wait for it |
|
|
|
thread.join() |
|
|
|
|
|
|
|
def geturl(path): |
|
|
|
req = urllib2.urlopen("http://127.0.0.1:12380" + path) |
|
|
|
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.server = nilmdb.Server(testdb, host = "127.0.0.1", |
|
|
|
port = 12380, stoppable = False) |
|
|
|
self.server.start(blocking = False) |
|
|
|
|
|
|
|
def tearDown(self): |
|
|
|
# Close web app |
|
|
|
self.server.stop() |
|
|
|
|
|
|
|
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(urllib2.HTTPError) as e: |
|
|
|
geturl(url) |
|
|
|
eq_(e.exception.code, 404) |
|
|
|
|
|
|
|
# Check version |
|
|
|
eq_(V(getjson("/version")), V(self.server.version)) |
|
|
|
|
|
|
|
def test_stream(self): |
|
|
|
# List |
|
|
|
streams = getjson("/stream/list") |
|
|
|
print streams |