progress_stream_test.py 1.2 KB

123456789101112131415161718192021222324252627282930313233343536
  1. from __future__ import absolute_import
  2. from __future__ import unicode_literals
  3. from six import StringIO
  4. from compose import progress_stream
  5. from tests import unittest
  6. class ProgressStreamTestCase(unittest.TestCase):
  7. def test_stream_output(self):
  8. output = [
  9. '{"status": "Downloading", "progressDetail": {"current": '
  10. '31019763, "start": 1413653874, "total": 62763875}, '
  11. '"progress": "..."}',
  12. ]
  13. events = progress_stream.stream_output(output, StringIO())
  14. self.assertEqual(len(events), 1)
  15. def test_stream_output_div_zero(self):
  16. output = [
  17. '{"status": "Downloading", "progressDetail": {"current": '
  18. '0, "start": 1413653874, "total": 0}, '
  19. '"progress": "..."}',
  20. ]
  21. events = progress_stream.stream_output(output, StringIO())
  22. self.assertEqual(len(events), 1)
  23. def test_stream_output_null_total(self):
  24. output = [
  25. '{"status": "Downloading", "progressDetail": {"current": '
  26. '0, "start": 1413653874, "total": null}, '
  27. '"progress": "..."}',
  28. ]
  29. events = progress_stream.stream_output(output, StringIO())
  30. self.assertEqual(len(events), 1)