sdk/python/tests/test_crunch.py

   1 import arvados.crunch
   2 import os
   3 import shutil
   4 import tempfile
   5 import unittest
   6
   7 class TaskOutputDirTest(unittest.TestCase):
   8     def setUp(self):
   9         self.tmp = tempfile.mkdtemp()
  10         os.environ['TASK_KEEPMOUNT_TMP'] = self.tmp
  11
  12     def tearDown(self):
  13         os.environ.pop('TASK_KEEPMOUNT_TMP')
  14         shutil.rmtree(self.tmp)
  15
  16     def test_env_var(self):
  17         out = arvados.crunch.TaskOutputDir()
  18         self.assertEqual(out.path, self.tmp)
  19
  20         with open(os.path.join(self.tmp, '.arvados#collection'), 'w') as f:
  21             f.write('{\n  "manifest_text":"",\n  "uuid":null\n}\n')
  22         self.assertEqual(out.manifest_text(), '')
  23
  24         # Special file must be re-read on each call to manifest_text().
  25         with open(os.path.join(self.tmp, '.arvados#collection'), 'w') as f:
  26             f.write(r'{"manifest_text":". unparsed 0:3:foo\n","uuid":null}')
  27         self.assertEqual(out.manifest_text(), ". unparsed 0:3:foo\n")