Jim Paris
84e21ff467
The server buffers the string and passes it to nilmdb. Nilmdb passes the string to bulkdata. Bulkdata uses the rocket interface to parse it in chunks, as necessary. Everything gets passed back up and everyone is happy. Currently, only pyrocket implements append_string.
50 lines
1.2 KiB
Python
50 lines
1.2 KiB
Python
# Just some helpers for test functions
|
|
|
|
import shutil, os
|
|
|
|
def myrepr(x):
|
|
if isinstance(x, basestring):
|
|
return '"' + x + '"'
|
|
else:
|
|
return repr(x)
|
|
|
|
def eq_(a, b):
|
|
if not a == b:
|
|
raise AssertionError("%s != %s" % (myrepr(a), myrepr(b)))
|
|
|
|
def lt_(a, b):
|
|
if not a < b:
|
|
raise AssertionError("%s is not less than %s" % (myrepr(a), myrepr(b)))
|
|
|
|
def in_(a, b):
|
|
if a not in b:
|
|
raise AssertionError("%s not in %s" % (myrepr(a), myrepr(b)))
|
|
|
|
def in2_(a1, a2, b):
|
|
if a1 not in b and a2 not in b:
|
|
raise AssertionError("(%s or %s) not in %s" % (myrepr(a1), myrepr(a2),
|
|
myrepr(b)))
|
|
|
|
def ne_(a, b):
|
|
if not a != b:
|
|
raise AssertionError("unexpected %s == %s" % (myrepr(a), myrepr(b)))
|
|
|
|
def lines_(a, n):
|
|
l = a.count('\n')
|
|
if not l == n:
|
|
if len(a) > 5000:
|
|
a = a[0:5000] + " ... truncated"
|
|
raise AssertionError("wanted %d lines, got %d in output: '%s'"
|
|
% (n, l, a))
|
|
|
|
def recursive_unlink(path):
|
|
try:
|
|
shutil.rmtree(path)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
os.unlink(path)
|
|
except OSError:
|
|
pass
|
|
|