sdk/python/tests/test_stream.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4
   5 import arvados
   6 from arvados import StreamReader, StreamFileReader
   7
   8 import run_test_server
   9
  10 class StreamReaderTestCase(unittest.TestCase):
  11     API_COLLECTIONS = run_test_server.fixture('collections')
  12
  13     def manifest_for(self, coll_name):
  14         return self.API_COLLECTIONS[coll_name]['manifest_text']
  15
  16     def test_manifest_text_without_keep_client(self):
  17         mtext = self.manifest_for('multilevel_collection_1')
  18         for line in mtext.rstrip('\n').split('\n'):
  19             reader = StreamReader(line.split())
  20             self.assertEqual(line + '\n', reader.manifest_text())
  21
  22
  23 if __name__ == '__main__':
  24     unittest.main()