Jim Paris
8e79998e95
Enable the following pragmas: synchronous=NORMAL, journal_mode=WAL. This offers a significant speedup to INSERT times compared to synchronous=FULL, and is roughly the same as synchronous=OFF but should be a bit safer.
257 lines
9.1 KiB
Python
257 lines
9.1 KiB
Python
import nilmdb.server
|
|
|
|
from nose.tools import *
|
|
from nose.tools import assert_raises
|
|
import distutils.version
|
|
import simplejson as json
|
|
import itertools
|
|
import os
|
|
import sys
|
|
import threading
|
|
import urllib2
|
|
from urllib2 import urlopen, HTTPError
|
|
import cStringIO
|
|
import time
|
|
import requests
|
|
|
|
from nilmdb.utils import serializer_proxy
|
|
|
|
testdb = "tests/testdb"
|
|
|
|
#@atexit.register
|
|
#def cleanup():
|
|
# os.unlink(testdb)
|
|
|
|
from testutil.helpers import *
|
|
|
|
class Test00Nilmdb(object): # named 00 so it runs first
|
|
def test_NilmDB(self):
|
|
recursive_unlink(testdb)
|
|
|
|
with assert_raises(IOError):
|
|
nilmdb.server.NilmDB("/nonexistant-db/foo")
|
|
|
|
db = nilmdb.server.NilmDB(testdb)
|
|
db.close()
|
|
db = nilmdb.server.NilmDB(testdb)
|
|
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.server.NilmDB(testdb)
|
|
eq_(db.stream_list(), [])
|
|
|
|
# Bad path
|
|
with assert_raises(ValueError):
|
|
db.stream_create("foo/bar/baz", "float32_8")
|
|
with assert_raises(ValueError):
|
|
db.stream_create("/foo", "float32_8")
|
|
# Bad layout type
|
|
with assert_raises(ValueError):
|
|
db.stream_create("/newton/prep", "NoSuchLayout")
|
|
db.stream_create("/newton/prep", "float32_8")
|
|
db.stream_create("/newton/raw", "uint16_6")
|
|
db.stream_create("/newton/zzz/rawnotch", "uint16_9")
|
|
|
|
# Verify we got 3 streams
|
|
eq_(db.stream_list(), [ ["/newton/prep", "float32_8"],
|
|
["/newton/raw", "uint16_6"],
|
|
["/newton/zzz/rawnotch", "uint16_9"]
|
|
])
|
|
# Match just one type or one path
|
|
eq_(db.stream_list(layout="uint16_6"), [ ["/newton/raw", "uint16_6"] ])
|
|
eq_(db.stream_list(path="/newton/raw"), [ ["/newton/raw", "uint16_6"] ])
|
|
|
|
# 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)
|
|
|
|
# fill in some test coverage for start >= end
|
|
with assert_raises(nilmdb.server.NilmDBError):
|
|
db.stream_remove("/newton/prep", 0, 0)
|
|
with assert_raises(nilmdb.server.NilmDBError):
|
|
db.stream_remove("/newton/prep", 1, 0)
|
|
db.stream_remove("/newton/prep", 0, 1)
|
|
|
|
db.close()
|
|
|
|
class TestBlockingServer(object):
|
|
def setUp(self):
|
|
self.db = serializer_proxy(nilmdb.server.NilmDB)(testdb)
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
|
|
def test_blocking_server(self):
|
|
# Server should fail if the database doesn't have a "_thread_safe"
|
|
# property.
|
|
with assert_raises(KeyError):
|
|
nilmdb.server.Server(object())
|
|
|
|
# Start web app on a custom port
|
|
self.server = nilmdb.server.Server(self.db, host = "127.0.0.1",
|
|
port = 32180, 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()
|
|
if not event.wait(timeout = 10):
|
|
raise AssertionError("server didn't start in 10 seconds")
|
|
|
|
# Send request to exit.
|
|
req = urlopen("http://127.0.0.1:32180/exit/", timeout = 1)
|
|
|
|
# Wait for it
|
|
thread.join()
|
|
|
|
def geturl(path):
|
|
req = urlopen("http://127.0.0.1:32180" + 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 = serializer_proxy(nilmdb.server.NilmDB)(testdb)
|
|
self.server = nilmdb.server.Server(self.db, host = "127.0.0.1",
|
|
port = 32180, 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.LooseVersion(getjson("/version")),
|
|
distutils.version.LooseVersion(nilmdb.__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', 'float32_8'],
|
|
['/newton/raw', 'uint16_6'],
|
|
['/newton/zzz/rawnotch', 'uint16_9'],
|
|
])
|
|
|
|
streams = getjson("/stream/list?layout=uint16_6")
|
|
eq_(streams, [['/newton/raw', 'uint16_6']])
|
|
|
|
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_cors_headers(self):
|
|
# Test that CORS headers are being set correctly
|
|
|
|
# Normal GET should send simple response
|
|
url = "http://127.0.0.1:32180/stream/list"
|
|
r = requests.get(url, headers = { "Origin": "http://google.com/" })
|
|
eq_(r.status_code, 200)
|
|
if "access-control-allow-origin" not in r.headers:
|
|
raise AssertionError("No Access-Control-Allow-Origin (CORS) "
|
|
"header in response:\n", r.headers)
|
|
eq_(r.headers["access-control-allow-origin"], "http://google.com/")
|
|
|
|
# OPTIONS without CORS preflight headers should result in 405
|
|
r = requests.options(url, headers = {
|
|
"Origin": "http://google.com/",
|
|
})
|
|
eq_(r.status_code, 405)
|
|
|
|
# OPTIONS with preflight headers should give preflight response
|
|
r = requests.options(url, headers = {
|
|
"Origin": "http://google.com/",
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "X-Custom",
|
|
})
|
|
eq_(r.status_code, 200)
|
|
if "access-control-allow-origin" not in r.headers:
|
|
raise AssertionError("No Access-Control-Allow-Origin (CORS) "
|
|
"header in response:\n", r.headers)
|
|
eq_(r.headers["access-control-allow-methods"], "GET, HEAD")
|
|
eq_(r.headers["access-control-allow-headers"], "X-Custom")
|
|
|
|
def test_post_bodies(self):
|
|
# Test JSON post bodies
|
|
r = requests.post("http://127.0.0.1:32180/stream/set_metadata",
|
|
headers = { "Content-Type": "application/json" },
|
|
data = '{"hello": 1}')
|
|
eq_(r.status_code, 404) # wrong parameters
|
|
|
|
r = requests.post("http://127.0.0.1:32180/stream/set_metadata",
|
|
headers = { "Content-Type": "application/json" },
|
|
data = '["hello"]')
|
|
eq_(r.status_code, 415) # not a dict
|
|
|
|
r = requests.post("http://127.0.0.1:32180/stream/set_metadata",
|
|
headers = { "Content-Type": "application/json" },
|
|
data = '[hello]')
|
|
eq_(r.status_code, 400) # badly formatted JSON
|