test_disk.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980
  1. """
  2. Unit tests for the disk utilities.
  3. """
  4. # Authors: Gael Varoquaux <gael dot varoquaux at normalesup dot org>
  5. # Lars Buitinck
  6. # Copyright (c) 2010 Gael Varoquaux
  7. # License: BSD Style, 3 clauses.
  8. from __future__ import with_statement
  9. import array
  10. import os
  11. from joblib.disk import disk_used, memstr_to_bytes, mkdirp, rm_subdirs
  12. from joblib.testing import parametrize, raises
  13. ###############################################################################
  14. def test_disk_used(tmpdir):
  15. cachedir = tmpdir.strpath
  16. # Not write a file that is 1M big in this directory, and check the
  17. # size. The reason we use such a big file is that it makes us robust
  18. # to errors due to block allocation.
  19. a = array.array("i")
  20. sizeof_i = a.itemsize
  21. target_size = 1024
  22. n = int(target_size * 1024 / sizeof_i)
  23. a = array.array("i", n * (1,))
  24. with open(os.path.join(cachedir, "test"), "wb") as output:
  25. a.tofile(output)
  26. assert disk_used(cachedir) >= target_size
  27. assert disk_used(cachedir) < target_size + 12
  28. @parametrize(
  29. "text,value",
  30. [
  31. ("80G", 80 * 1024**3),
  32. ("1.4M", int(1.4 * 1024**2)),
  33. ("120M", 120 * 1024**2),
  34. ("53K", 53 * 1024),
  35. ],
  36. )
  37. def test_memstr_to_bytes(text, value):
  38. assert memstr_to_bytes(text) == value
  39. @parametrize(
  40. "text,exception,regex",
  41. [
  42. ("fooG", ValueError, r"Invalid literal for size.*fooG.*"),
  43. ("1.4N", ValueError, r"Invalid literal for size.*1.4N.*"),
  44. ],
  45. )
  46. def test_memstr_to_bytes_exception(text, exception, regex):
  47. with raises(exception) as excinfo:
  48. memstr_to_bytes(text)
  49. assert excinfo.match(regex)
  50. def test_mkdirp(tmpdir):
  51. mkdirp(os.path.join(tmpdir.strpath, "ham"))
  52. mkdirp(os.path.join(tmpdir.strpath, "ham"))
  53. mkdirp(os.path.join(tmpdir.strpath, "spam", "spam"))
  54. # Not all OSErrors are ignored
  55. with raises(OSError):
  56. mkdirp("")
  57. def test_rm_subdirs(tmpdir):
  58. sub_path = os.path.join(tmpdir.strpath, "subdir_one", "subdir_two")
  59. full_path = os.path.join(sub_path, "subdir_three")
  60. mkdirp(os.path.join(full_path))
  61. rm_subdirs(sub_path)
  62. assert os.path.exists(sub_path)
  63. assert not os.path.exists(full_path)