Previous commits went back and forth a bit on whether the various APIs should use bytes or strings, but bytes appears to be a better answer, because actual data in streams will always be 7-bit ASCII or raw binary. There's no reason to apply the performance penalty of constantly converting between bytes and strings. One drawback now is that lots of code now has to have "b" prefixes on strings, especially in tests, which inflates this commit quite a bit.
88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
import nilmdb
|
|
from nilmdb.utils.printf import *
|
|
import datetime_tz
|
|
|
|
from nose.tools import *
|
|
from nose.tools import assert_raises
|
|
import os
|
|
import sys
|
|
import io
|
|
|
|
from testutil.helpers import *
|
|
|
|
from nilmdb.utils import timestamper
|
|
|
|
class TestTimestamper(object):
|
|
|
|
# Not a very comprehensive test, but it's good enough.
|
|
|
|
def test_timestamper(self):
|
|
def join(list):
|
|
return b"\n".join(list) + b"\n"
|
|
|
|
datetime_tz.localtz_set("America/New_York")
|
|
|
|
start = nilmdb.utils.time.parse_time("03/24/2012")
|
|
lines_in = [ b"hello", b"world", b"hello world", b"# commented out" ]
|
|
lines_out = [ b"1332561600000000 hello",
|
|
b"1332561600000125 world",
|
|
b"1332561600000250 hello world" ]
|
|
|
|
# full
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperRate(input, start, 8000)
|
|
foo = ts.readlines()
|
|
eq_(foo, join(lines_out))
|
|
in_("TimestamperRate(..., start=", str(ts))
|
|
|
|
# first 30 or so bytes means the first 2 lines
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperRate(input, start, 8000)
|
|
foo = ts.readlines(30)
|
|
eq_(foo, join(lines_out[0:2]))
|
|
|
|
# stop iteration early
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperRate(input, start, 8000,
|
|
1332561600000200)
|
|
foo = b""
|
|
for line in ts:
|
|
foo += line
|
|
eq_(foo, join(lines_out[0:2]))
|
|
|
|
# stop iteration early (readlines)
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperRate(input, start, 8000,
|
|
1332561600000200)
|
|
foo = ts.readlines()
|
|
eq_(foo, join(lines_out[0:2]))
|
|
|
|
# stop iteration really early
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperRate(input, start, 8000,
|
|
1332561600000000)
|
|
foo = ts.readlines()
|
|
eq_(foo, b"")
|
|
|
|
# use iterator
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperRate(input, start, 8000)
|
|
foo = b""
|
|
for line in ts:
|
|
foo += line
|
|
eq_(foo, join(lines_out))
|
|
|
|
# check that TimestamperNow gives similar result
|
|
input = io.BytesIO(join(lines_in))
|
|
ts = timestamper.TimestamperNow(input)
|
|
foo = ts.readlines()
|
|
ne_(foo, join(lines_out))
|
|
eq_(len(foo), len(join(lines_out)))
|
|
eq_(str(ts), "TimestamperNow(...)")
|
|
|
|
# Test passing a file (should be empty)
|
|
ts = timestamper.TimestamperNow("/dev/null")
|
|
for line in ts:
|
|
raise AssertionError
|
|
ts.close()
|