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.
 
 
 

113 lines
3.7 KiB

  1. from nose.tools import *
  2. from nose.tools import assert_raises
  3. from testutil.helpers import *
  4. import io
  5. import os
  6. import sys
  7. import time
  8. import nilmdb.server
  9. from nilmdb.utils import timer, lock
  10. class TestMisc(object):
  11. def test_timer(self):
  12. capture = io.StringIO()
  13. old = sys.stdout
  14. sys.stdout = capture
  15. with nilmdb.utils.Timer("test"):
  16. time.sleep(0.01)
  17. with nilmdb.utils.Timer("test syslog", tosyslog=True):
  18. time.sleep(0.01)
  19. sys.stdout = old
  20. in_("test: ", capture.getvalue())
  21. def test_lock(self):
  22. with open("/dev/null") as f:
  23. eq_(nilmdb.utils.lock.exclusive_lock(f), True)
  24. nilmdb.utils.lock.exclusive_unlock(f)
  25. # Test error conditions
  26. class FakeFile():
  27. def __init__(self, fileno):
  28. self._fileno = fileno
  29. def fileno(self):
  30. return self._fileno
  31. with assert_raises(TypeError):
  32. nilmdb.utils.lock.exclusive_lock(FakeFile('none'))
  33. with assert_raises(ValueError):
  34. nilmdb.utils.lock.exclusive_lock(FakeFile(-1))
  35. with assert_raises(IOError):
  36. nilmdb.utils.lock.exclusive_lock(FakeFile(12345))
  37. # Lock failure is tested in test_bulkdata
  38. def test_replace_file(self):
  39. fn = b"tests/misc-testdb/file"
  40. try:
  41. os.mkdir(os.path.dirname(fn))
  42. except FileExistsError:
  43. pass
  44. with open(fn, "wb") as f:
  45. f.write(b"hello, world")
  46. nilmdb.utils.atomic.replace_file(fn, b"goodbye, world")
  47. with open(fn, "rb") as f:
  48. eq_(f.read(), b"goodbye, world")
  49. def test_punch(self):
  50. fn = b"tests/misc-testdb/punchit"
  51. try:
  52. os.mkdir(os.path.dirname(fn))
  53. except FileExistsError:
  54. pass
  55. with open(fn, "wb") as f:
  56. f.write(b"hello, world")
  57. nilmdb.utils.fallocate.punch_hole(fn, 3, 5)
  58. with open(fn, "rb") as f:
  59. eq_(f.read(), b"hel\0\0\0\0\0orld")
  60. with assert_raises(OSError):
  61. nilmdb.utils.fallocate.punch_hole(fn, 1, -1, False)
  62. with assert_raises(OSError):
  63. nilmdb.utils.fallocate.punch_hole("/", 1, 1, False)
  64. # no exception because we ignore errors by default
  65. nilmdb.utils.fallocate.punch_hole(fn, 1, -1)
  66. def test_diskusage(self):
  67. hs = nilmdb.utils.diskusage.human_size
  68. eq_(hs(0), "0 bytes")
  69. eq_(hs(1), "1 byte")
  70. eq_(hs(1023), "1023 bytes")
  71. eq_(hs(1024), "1 kiB")
  72. eq_(hs(1048575), "1024 kiB")
  73. eq_(hs(1048576), "1.0 MiB")
  74. eq_(hs(1073741823), "1024.0 MiB")
  75. eq_(hs(1073741824), "1.00 GiB")
  76. eq_(hs(1099511627775), "1024.00 GiB")
  77. eq_(hs(1099511627776), "1.00 TiB")
  78. eq_(hs(1099511627776 * 5000.1234), "5000.12 TiB")
  79. nilmdb.utils.diskusage.du("/dev")
  80. with assert_raises(OSError):
  81. nilmdb.utils.diskusage.du("/dev/null/bogus")
  82. nilmdb.utils.diskusage.du("super-bogus-does-not-exist")
  83. def tests_cors_allow(self):
  84. # Just to get some test coverage; these code paths aren't actually
  85. # used in current code
  86. cpy = nilmdb.server.serverutil.cherrypy
  87. (req, resp) = (cpy.request, cpy.response)
  88. cpy.request.method = "DELETE"
  89. with assert_raises(cpy.HTTPError):
  90. nilmdb.server.serverutil.CORS_allow(methods="POST")
  91. with assert_raises(cpy.HTTPError):
  92. nilmdb.server.serverutil.CORS_allow(methods=["POST"])
  93. with assert_raises(cpy.HTTPError):
  94. nilmdb.server.serverutil.CORS_allow(methods=["GET"])
  95. with assert_raises(cpy.HTTPError):
  96. nilmdb.server.serverutil.CORS_allow(methods=[])
  97. (cpy.request, cpy.response) = (req, resp)