You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

88 lines
2.7 KiB

  1. import nilmdb
  2. from nilmdb.utils.printf import *
  3. import datetime_tz
  4. from nose.tools import *
  5. from nose.tools import assert_raises
  6. import os
  7. import sys
  8. import io
  9. from testutil.helpers import *
  10. from nilmdb.utils import timestamper
  11. class TestTimestamper(object):
  12. # Not a very comprehensive test, but it's good enough.
  13. def test_timestamper(self):
  14. def join(list):
  15. return "\n".join(list) + "\n"
  16. datetime_tz.localtz_set("America/New_York")
  17. start = nilmdb.utils.time.parse_time("03/24/2012")
  18. lines_in = [ "hello", "world", "hello world", "# commented out" ]
  19. lines_out = [ "1332561600000000 hello",
  20. "1332561600000125 world",
  21. "1332561600000250 hello world" ]
  22. # full
  23. input = io.StringIO(join(lines_in))
  24. ts = timestamper.TimestamperRate(input, start, 8000)
  25. foo = ts.readlines()
  26. eq_(foo, join(lines_out))
  27. in_("TimestamperRate(..., start=", str(ts))
  28. # first 30 or so bytes means the first 2 lines
  29. input = io.StringIO(join(lines_in))
  30. ts = timestamper.TimestamperRate(input, start, 8000)
  31. foo = ts.readlines(30)
  32. eq_(foo, join(lines_out[0:2]))
  33. # stop iteration early
  34. input = io.StringIO(join(lines_in))
  35. ts = timestamper.TimestamperRate(input, start, 8000,
  36. 1332561600000200)
  37. foo = ""
  38. for line in ts:
  39. foo += line
  40. eq_(foo, join(lines_out[0:2]))
  41. # stop iteration early (readlines)
  42. input = io.StringIO(join(lines_in))
  43. ts = timestamper.TimestamperRate(input, start, 8000,
  44. 1332561600000200)
  45. foo = ts.readlines()
  46. eq_(foo, join(lines_out[0:2]))
  47. # stop iteration really early
  48. input = io.StringIO(join(lines_in))
  49. ts = timestamper.TimestamperRate(input, start, 8000,
  50. 1332561600000000)
  51. foo = ts.readlines()
  52. eq_(foo, "")
  53. # use iterator
  54. input = io.StringIO(join(lines_in))
  55. ts = timestamper.TimestamperRate(input, start, 8000)
  56. foo = ""
  57. for line in ts:
  58. foo += line
  59. eq_(foo, join(lines_out))
  60. # check that TimestamperNow gives similar result
  61. input = io.StringIO(join(lines_in))
  62. ts = timestamper.TimestamperNow(input)
  63. foo = ts.readlines()
  64. ne_(foo, join(lines_out))
  65. eq_(len(foo), len(join(lines_out)))
  66. eq_(str(ts), "TimestamperNow(...)")
  67. # Test passing a file (should be empty)
  68. ts = timestamper.TimestamperNow("/dev/null")
  69. for line in ts:
  70. raise AssertionError
  71. ts.close()