nilmdb/tests/test_timestamper.py
Jim Paris e1bf680d13 pass data to server
git-svn-id: https://bucket.mit.edu/svn/nilm/nilmdb@10655 ddd99763-3ecb-0310-9145-efcb8ce7c51f
2012-03-26 22:28:33 +00:00

89 lines
2.8 KiB
Python

import nilmdb
from nilmdb.printf import *
import datetime_tz
from nose.tools import *
from nose.tools import assert_raises
import os
import sys
import cStringIO
def eq_(a, b):
if not a == b:
raise AssertionError("%r != %r" % (a, b))
def ne_(a, b):
if not a != b:
raise AssertionError("unexpected %r == %r" % (a, b))
class TestTimestamper(object):
# Not a very comprehensive test, but it's good enough.
def test_timestamper(self):
def join(list):
return "\n".join(list) + "\n"
start = datetime_tz.datetime_tz.smartparse("03/24/2012").totimestamp()
lines_in = [ "hello", "world", "hello world", "# commented out" ]
lines_out = [ "1332561600.000000 hello",
"1332561600.000125 world",
"1332561600.000250 hello world" ]
# full
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperRate(input, start, 8000)
foo = ts.readlines()
eq_(foo, join(lines_out))
# first 30 or so bytes means the first 2 lines
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperRate(input, start, 8000)
foo = ts.readlines(30)
eq_(foo, join(lines_out[0:2]))
# stop iteration early
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperRate(input, start, 8000,
1332561600.000200)
foo = ""
for line in ts:
foo += line
eq_(foo, join(lines_out[0:2]))
# stop iteration early (readlines)
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperRate(input, start, 8000,
1332561600.000200)
foo = ts.readlines()
eq_(foo, join(lines_out[0:2]))
# stop iteration really early
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperRate(input, start, 8000,
1332561600.000000)
foo = ts.readlines()
eq_(foo, "")
# use iterator
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperRate(input, start, 8000)
foo = ""
for line in ts:
foo += line
eq_(foo, join(lines_out))
# check that TimestamperNow gives similar result
input = cStringIO.StringIO(join(lines_in))
ts = nilmdb.timestamper.TimestamperNow(input)
foo = ts.readlines()
ne_(foo, join(lines_out))
eq_(len(foo), len(join(lines_out)))
# Test passing a file (should be empty)
ts = nilmdb.timestamper.TimestamperNow("/dev/null")
for line in ts:
raise AssertionError
ts.close()