sdk/python/tests/test_arv_keepdocker.py

   1 from __future__ import absolute_import
   2 import arvados
   3 import hashlib
   4 import mock
   5 import os
   6 import subprocess
   7 import sys
   8 import tempfile
   9 import unittest
  10 import logging
  11
  12 import arvados.commands.keepdocker as arv_keepdocker
  13 from . import arvados_testutil as tutil
  14 from . import run_test_server
  15
  16
  17 class StopTest(Exception):
  18     pass
  19
  20
  21 class ArvKeepdockerTestCase(unittest.TestCase, tutil.VersionChecker):
  22     def run_arv_keepdocker(self, args, err):
  23         sys.argv = ['arv-keepdocker'] + args
  24         log_handler = logging.StreamHandler(err)
  25         arv_keepdocker.logger.addHandler(log_handler)
  26         try:
  27             return arv_keepdocker.main()
  28         finally:
  29             arv_keepdocker.logger.removeHandler(log_handler)
  30
  31     def test_unsupported_arg(self):
  32         with self.assertRaises(SystemExit):
  33             self.run_arv_keepdocker(['-x=unknown'], sys.stderr)
  34
  35     def test_version_argument(self):
  36         with tutil.redirected_streams(
  37                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  38             with self.assertRaises(SystemExit):
  39                 self.run_arv_keepdocker(['--version'], sys.stderr)
  40         self.assertVersionOutput(out, err)
  41
  42     @mock.patch('arvados.commands.keepdocker.find_image_hashes',
  43                 return_value=['abc123'])
  44     @mock.patch('arvados.commands.keepdocker.find_one_image_hash',
  45                 return_value='abc123')
  46     def test_image_format_compatibility(self, _1, _2):
  47         old_id = hashlib.sha256(b'old').hexdigest()
  48         new_id = 'sha256:'+hashlib.sha256(b'new').hexdigest()
  49         for supported, img_id, expect_ok in [
  50                 (['v1'], old_id, True),
  51                 (['v1'], new_id, False),
  52                 (None, old_id, False),
  53                 ([], old_id, False),
  54                 ([], new_id, False),
  55                 (['v1', 'v2'], new_id, True),
  56                 (['v1'], new_id, False),
  57                 (['v2'], new_id, True)]:
  58
  59             fakeDD = arvados.api('v1')._rootDesc
  60             if supported is None:
  61                 del fakeDD['dockerImageFormats']
  62             else:
  63                 fakeDD['dockerImageFormats'] = supported
  64
  65             err = tutil.StringIO()
  66             out = tutil.StringIO()
  67
  68             with tutil.redirected_streams(stdout=out), \
  69                  mock.patch('arvados.api') as api, \
  70                  mock.patch('arvados.commands.keepdocker.popen_docker',
  71                             return_value=subprocess.Popen(
  72                                 ['echo', img_id],
  73                                 stdout=subprocess.PIPE)), \
  74                  mock.patch('arvados.commands.keepdocker.prep_image_file',
  75                             side_effect=StopTest), \
  76                  self.assertRaises(StopTest if expect_ok else SystemExit):
  77
  78                 api()._rootDesc = fakeDD
  79                 self.run_arv_keepdocker(['--force', 'testimage'], err)
  80
  81             self.assertEqual(out.getvalue(), '')
  82             if expect_ok:
  83                 self.assertNotRegex(
  84                     err.getvalue(), "refusing to store",
  85                     msg=repr((supported, img_id)))
  86             else:
  87                 self.assertRegex(
  88                     err.getvalue(), "refusing to store",
  89                     msg=repr((supported, img_id)))
  90             if not supported:
  91                 self.assertRegex(
  92                     err.getvalue(),
  93                     "server does not specify supported image formats",
  94                     msg=repr((supported, img_id)))
  95
  96         fakeDD = arvados.api('v1')._rootDesc
  97         fakeDD['dockerImageFormats'] = ['v1']
  98         err = tutil.StringIO()
  99         out = tutil.StringIO()
 100         with tutil.redirected_streams(stdout=out), \
 101              mock.patch('arvados.api') as api, \
 102              mock.patch('arvados.commands.keepdocker.popen_docker',
 103                         return_value=subprocess.Popen(
 104                             ['echo', new_id],
 105                             stdout=subprocess.PIPE)), \
 106              mock.patch('arvados.commands.keepdocker.prep_image_file',
 107                         side_effect=StopTest), \
 108              self.assertRaises(StopTest):
 109             api()._rootDesc = fakeDD
 110             self.run_arv_keepdocker(
 111                 ['--force', '--force-image-format', 'testimage'], err)
 112         self.assertRegex(err.getvalue(), "forcing incompatible image")