This is so tests can be run without interfering with a normal server.tags/nilmdb-1.2.2
@@ -25,7 +25,7 @@ import re | |||
from testutil.helpers import * | |||
testdb = "tests/client-testdb" | |||
testurl = "http://localhost:12380/" | |||
testurl = "http://localhost:32180/" | |||
def setup_module(): | |||
global test_server, test_db | |||
@@ -35,7 +35,7 @@ def setup_module(): | |||
# Start web app on a custom port | |||
test_db = nilmdb.utils.serializer_proxy(nilmdb.NilmDB)(testdb, sync = False) | |||
test_server = nilmdb.Server(test_db, host = "127.0.0.1", | |||
port = 12380, stoppable = False, | |||
port = 32180, stoppable = False, | |||
fast_shutdown = True, | |||
force_traceback = False) | |||
test_server.start(blocking = False) | |||
@@ -62,13 +62,13 @@ class TestClient(object): | |||
client.close() | |||
# Then a fake URL on a real host | |||
client = nilmdb.Client(url = "http://localhost:12380/fake/") | |||
client = nilmdb.Client(url = "http://localhost:32180/fake/") | |||
with assert_raises(nilmdb.client.ClientError): | |||
client.version() | |||
client.close() | |||
# Now a real URL with no http:// prefix | |||
client = nilmdb.Client(url = "localhost:12380") | |||
client = nilmdb.Client(url = "localhost:32180") | |||
version = client.version() | |||
client.close() | |||
@@ -584,7 +584,7 @@ class TestClient(object): | |||
def connections(): | |||
try: | |||
poolmanager = c.http._last_response.connection.poolmanager | |||
pool = poolmanager.pools[('http','localhost',12380)] | |||
pool = poolmanager.pools[('http','localhost',32180)] | |||
return (pool.num_connections, pool.num_requests) | |||
except: | |||
raise SkipTest("can't get connection info") | |||
@@ -32,7 +32,7 @@ def server_start(max_results = None, bulkdata_args = {}): | |||
max_results = max_results, | |||
bulkdata_args = bulkdata_args) | |||
test_server = nilmdb.Server(test_db, host = "127.0.0.1", | |||
port = 12380, stoppable = False, | |||
port = 32180, stoppable = False, | |||
fast_shutdown = True, | |||
force_traceback = False) | |||
test_server.start(blocking = False) | |||
@@ -64,6 +64,7 @@ class TestCmdline(object): | |||
passing the given input. Returns a tuple with the output and | |||
exit code""" | |||
# printf("TZ=UTC ./nilmtool.py %s\n", arg_string) | |||
os.environ['NILMDB_URL'] = "http://localhost:32180/" | |||
class stdio_wrapper: | |||
def __init__(self, stdin, stdout, stderr): | |||
self.io = (stdin, stdout, stderr) | |||
@@ -174,7 +175,7 @@ class TestCmdline(object): | |||
self.fail("-u localhost:1 info") | |||
self.contain("error connecting to server") | |||
self.ok("-u localhost:12380 info") | |||
self.ok("-u localhost:32180 info") | |||
self.ok("info") | |||
# Duplicated arguments should fail, but this isn't implemented | |||
@@ -210,7 +211,7 @@ class TestCmdline(object): | |||
def test_03_info(self): | |||
self.ok("info") | |||
self.contain("Server URL: http://localhost:12380/") | |||
self.contain("Server URL: http://localhost:32180/") | |||
self.contain("Client version: " + nilmdb.__version__) | |||
self.contain("Server version: " + test_server.version) | |||
self.contain("Server database path") | |||
@@ -119,7 +119,7 @@ class TestBlockingServer(object): | |||
# Start web app on a custom port | |||
self.server = nilmdb.Server(self.db, host = "127.0.0.1", | |||
port = 12380, stoppable = True) | |||
port = 32180, stoppable = True) | |||
# Run it | |||
event = threading.Event() | |||
@@ -131,13 +131,13 @@ class TestBlockingServer(object): | |||
raise AssertionError("server didn't start in 10 seconds") | |||
# Send request to exit. | |||
req = urlopen("http://127.0.0.1:12380/exit/", timeout = 1) | |||
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:12380" + path, timeout = 10) | |||
req = urlopen("http://127.0.0.1:32180" + path, timeout = 10) | |||
return req.read() | |||
def getjson(path): | |||
@@ -149,7 +149,7 @@ class TestServer(object): | |||
# Start web app on a custom port | |||
self.db = serializer_proxy(nilmdb.NilmDB)(testdb, sync=False) | |||
self.server = nilmdb.Server(self.db, host = "127.0.0.1", | |||
port = 12380, stoppable = False) | |||
port = 32180, stoppable = False) | |||
self.server.start(blocking = False) | |||
def tearDown(self): | |||