nilmdb/tests/test_timestamper.py
Jim Paris 57751f5b32 Consistently use bytes everywhere for stream data
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.
2019-08-13 15:53:05 -04:00

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()