sdk/python/tests/test_arv_keepdocker.py

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