sdk/python/tests/test_crunch.py

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